From e1dbd1cafbe2a7018dac0d224eac456ad3a15998 Mon Sep 17 00:00:00 2001 From: shy3130 <415333856@qq.com> Date: Mon, 20 Jul 2026 10:33:06 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E7=9B=91=E6=8E=A7=E4=B8=AD?= =?UTF-8?q?=E5=BF=83=E6=B6=A8=E8=B7=8C=E5=81=9C=E4=BF=A1=E5=8F=B7=E8=AF=AF?= =?UTF-8?q?=E6=8A=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/indicators/pipeline.py | 48 +++++++++++++++----- backend/tests/test_price_limits.py | 70 ++++++++++++++++++++++++++++++ 2 files changed, 107 insertions(+), 11 deletions(-) diff --git a/backend/app/indicators/pipeline.py b/backend/app/indicators/pipeline.py index bad5a3a..210129b 100644 --- a/backend/app/indicators/pipeline.py +++ b/backend/app/indicators/pipeline.py @@ -677,6 +677,10 @@ def compute_limit_signals( continue if c in instruments.columns: inst_cols.append(c) + if need_price_limits and "as_of" in instruments.columns: + inst_cols.append( + pl.col("as_of").cast(pl.Date, strict=False).alias("_instrument_as_of") + ) inst_subset = instruments.select(inst_cols).unique(subset=["symbol"]) if need_price_limits and "name" in instruments.columns: @@ -741,19 +745,28 @@ def compute_limit_signals( .alias("_theoretical_limit_down") ) - # 生效涨跌停价: 最新日优先使用维表权威值; 历史日期继续使用理论价。 - # instruments 只有最新快照, 不能用于历史日期; >=10000 视为新股无涨跌停限制哨兵值。 + # 生效涨跌停价: 维表日期与行情日期一致时使用权威值, 否则使用理论价。 + # 旧版维表没有 as_of, 保持仅在最新行情日使用权威值的兼容行为。 _SENTINEL = 10000.0 - is_latest_date = pl.col("date") == pl.col("date").max() + if "_instrument_as_of" in df.columns: + authoritative_date = ( + pl.col("_instrument_as_of") == pl.col("date").cast(pl.Date, strict=False) + ) + else: + authoritative_date = pl.col("date") == pl.col("date").max() if "limit_up" in df.columns: effective_limit_up = pl.when( - is_latest_date & pl.col("limit_up").is_not_null() & (pl.col("limit_up") < _SENTINEL) + authoritative_date + & pl.col("limit_up").is_not_null() + & (pl.col("limit_up") < _SENTINEL) ).then(pl.col("limit_up")).otherwise(pl.col("_theoretical_limit_up")) else: effective_limit_up = pl.col("_theoretical_limit_up") if "limit_down" in df.columns: effective_limit_down = pl.when( - is_latest_date & pl.col("limit_down").is_not_null() & (pl.col("limit_down") < _SENTINEL) + authoritative_date + & pl.col("limit_down").is_not_null() + & (pl.col("limit_down") < _SENTINEL) ).then(pl.col("limit_down")).otherwise(pl.col("_theoretical_limit_down")) else: effective_limit_down = pl.col("_theoretical_limit_down") @@ -870,7 +883,7 @@ def compute_limit_signals( cleanup = ["_prev_raw_close", "_limit_pct", "_theoretical_limit_up", "_theoretical_limit_down", "_effective_limit_up", "_effective_limit_down", - "_grp_up", "_grp_down"] + "_grp_up", "_grp_down", "_instrument_as_of"] if "_is_st" in df.columns: cleanup.append("_is_st") # 清理 join 产生的重复列 @@ -1658,6 +1671,10 @@ def _compute_limit_signals_today(df: pl.DataFrame, instruments: pl.DataFrame) -> for c in ["float_shares", "limit_up", "limit_down"]: if c in instruments.columns: inst_cols.append(c) + if "as_of" in instruments.columns: + inst_cols.append( + pl.col("as_of").cast(pl.Date, strict=False).alias("_instrument_as_of") + ) inst_subset = instruments.select(inst_cols).unique(subset=["symbol"]) if "name" in instruments.columns: st_flag = ( @@ -1704,19 +1721,28 @@ def _compute_limit_signals_today(df: pl.DataFrame, instruments: pl.DataFrame) -> limit_up_price = polars_limit_price(prev_raw, limit_pct, up=True) limit_down_price = polars_limit_price(prev_raw, limit_pct, up=False) - # 生效涨跌停价: 优先用维表权威值 (instruments.limit_up/down, 交易所级别精确价), - # 维表缺失 (新股上市前 5 日: limit_up 为 null 或哨兵 100000) 回退自算理论价。 + # 生效涨跌停价: 维表日期与行情日期一致时优先使用交易所权威值; + # 维表过期、价格缺失或新股哨兵值均回退自算理论价。旧版无 as_of 维表保持兼容。 # 哨兵阈值 10000 用于识别 "新股无涨跌停限制" 的占位值 (实际涨停价不可能上万)。 _SENTINEL = 10000.0 + authoritative_date = ( + pl.col("_instrument_as_of") == trade_date.cast(pl.Date, strict=False) + if "_instrument_as_of" in df.columns + else pl.lit(True) + ) if "limit_up" in df.columns: effective_limit_up = pl.when( - pl.col("limit_up").is_not_null() & (pl.col("limit_up") < _SENTINEL) + authoritative_date + & pl.col("limit_up").is_not_null() + & (pl.col("limit_up") < _SENTINEL) ).then(pl.col("limit_up")).otherwise(limit_up_price) else: effective_limit_up = limit_up_price if "limit_down" in df.columns: effective_limit_down = pl.when( - pl.col("limit_down").is_not_null() & (pl.col("limit_down") < _SENTINEL) + authoritative_date + & pl.col("limit_down").is_not_null() + & (pl.col("limit_down") < _SENTINEL) ).then(pl.col("limit_down")).otherwise(limit_down_price) else: effective_limit_down = limit_down_price @@ -1772,7 +1798,7 @@ def _compute_limit_signals_today(df: pl.DataFrame, instruments: pl.DataFrame) -> ]) # 清理 - cleanup = ["_limit_pct", "_is_st", "limit_up", "limit_down"] + cleanup = ["_limit_pct", "_is_st", "limit_up", "limit_down", "_instrument_as_of"] for c in df.columns: if c.endswith("_inst"): cleanup.append(c) diff --git a/backend/tests/test_price_limits.py b/backend/tests/test_price_limits.py index 822ff51..218bf55 100644 --- a/backend/tests/test_price_limits.py +++ b/backend/tests/test_price_limits.py @@ -8,6 +8,7 @@ import pytest from app.api import kline from app.backtest.matrix import load_market_data_matrix_from_parquet +from app.indicators import pipeline from app.price_limits import ( numpy_limit_price, numpy_price_limit_matrix, @@ -145,3 +146,72 @@ def test_minute_price_limit_prefers_authoritative_prices_only_today(monkeypatch) "limit_down": None, "source": "rule", } + + +def _daily_limit_rows(current_close: float) -> pl.DataFrame: + return pl.DataFrame({ + "symbol": ["600001.SH", "600001.SH"], + "date": [date(2026, 7, 17), date(2026, 7, 20)], + "open": [10.0, current_close], + "high": [10.0, current_close], + "low": [10.0, current_close], + "close": [10.0, current_close], + "raw_close": [10.0, current_close], + "raw_high": [10.0, current_close], + }) + + +@pytest.mark.parametrize( + ("instrument_as_of", "expected"), + [ + (date(2026, 7, 17), False), + (date(2026, 7, 20), True), + (None, True), + ], +) +def test_daily_limit_prices_require_matching_instrument_date(instrument_as_of, expected): + instrument_data = { + "symbol": ["600001.SH"], + "name": ["普通股"], + "limit_up": [10.90], + "limit_down": [9.10], + } + if instrument_as_of is not None: + instrument_data["as_of"] = [instrument_as_of] + + result = pipeline.compute_limit_signals( + _daily_limit_rows(9.10), + pl.DataFrame(instrument_data), + needed={"signal_limit_down"}, + ) + + assert result["signal_limit_down"][-1] is expected + assert "_instrument_as_of" not in result.columns + + +def test_realtime_limit_prices_ignore_stale_instrument_date(): + today = date(2026, 7, 20) + rows = pl.DataFrame({ + "symbol": ["600001.SH"], + "date": [today], + "open": [9.10], + "high": [9.10], + "low": [9.10], + "close": [9.10], + "raw_close": [9.10], + "raw_high": [9.10], + "_prev_close_raw": [10.0], + "volume": [1000.0], + }) + instruments = pl.DataFrame({ + "symbol": ["600001.SH"], + "name": ["普通股"], + "limit_up": [10.90], + "limit_down": [9.10], + "as_of": [date(2026, 7, 17)], + }) + + result = pipeline._compute_limit_signals_today(rows, instruments) + + assert result["signal_limit_down"][0] is False + assert "_instrument_as_of" not in result.columns