95 lines
3.2 KiB
Python
95 lines
3.2 KiB
Python
"""A 股涨跌停纯函数（板块表 + 封板判断）。
|
|
|
|
全函数无副作用，全部用 **raw** 价格（spec §3.3 / §6.2）。
|
|
封板判据：
|
|
- 严格一字板：open=high=low=close=limit_price → 无对手盘
|
|
- T 字板（涨停）：开=涨停 收=涨停 low<open → 盘中砸过板，保守拒单
|
|
- T 字板（跌停）：开=跌停 收=跌停 high>open → 盘中反弹过，对称保守拒单
|
|
"""
|
|
import pandas as pd
|
|
|
|
|
|
# ---- 板块分类（按代码前缀）----
|
|
def get_board(symbol: str) -> str:
|
|
"""按 symbol 前缀判断板块（spec §6.2）。"""
|
|
if symbol.startswith(("300", "301")):
|
|
return "gem" # 创业板
|
|
if symbol.startswith(("688", "689")):
|
|
return "star" # 科创板
|
|
if symbol.startswith(("8", "4", "920")):
|
|
return "bse" # 北交所
|
|
return "main" # 沪/深主板
|
|
|
|
|
|
_LIMIT_RATIO = {"main": 0.10, "gem": 0.20, "star": 0.20, "bse": 0.30}
|
|
_ST_RATIO = 0.05
|
|
|
|
|
|
def limit_ratio(board: str, is_st: bool) -> float:
|
|
"""涨跌停幅度。ST 统一 5%（北交所 ST 同 30%）。"""
|
|
if is_st and board != "bse":
|
|
return _ST_RATIO
|
|
return _LIMIT_RATIO[board]
|
|
|
|
|
|
# ---- 涨跌停价（按 pricetick 四舍五入取整）----
|
|
def limit_up_price(prev_close_raw: float, ratio: float, pricetick: float) -> float:
|
|
return round(prev_close_raw * (1 + ratio) / pricetick) * pricetick
|
|
|
|
|
|
def limit_down_price(prev_close_raw: float, ratio: float, pricetick: float) -> float:
|
|
return round(prev_close_raw * (1 - ratio) / pricetick) * pricetick
|
|
|
|
|
|
# ---- 封板形态 ----
|
|
def is_one_word_lock(bar: pd.Series, limit_price: float) -> bool:
|
|
"""严格一字板：开=高=低=收=limit_price。"""
|
|
return bool(
|
|
bar["open"] == bar["high"] == bar["low"] == bar["close"] == limit_price
|
|
)
|
|
|
|
|
|
def is_t_lock(bar: pd.Series, limit_price: float) -> bool:
|
|
"""T 字板（涨停型）：开=涨停、收=涨停、low<open（盘中砸过涨停）。"""
|
|
return bool(
|
|
bar["open"] == limit_price
|
|
and bar["close"] == limit_price
|
|
and bar["low"] < bar["open"]
|
|
)
|
|
|
|
|
|
def _is_t_lock_down(bar: pd.Series, limit_price: float) -> bool:
|
|
"""T 字板（跌停型，对称）：开=跌停、收=跌停、high>open（盘中反弹过跌停）。"""
|
|
return bool(
|
|
bar["open"] == limit_price
|
|
and bar["close"] == limit_price
|
|
and bar["high"] > bar["open"]
|
|
)
|
|
|
|
|
|
# ---- 正式入口（带 symbol，自动判板块）----
|
|
def is_locked_for_buy_symbol(
|
|
bar: pd.Series,
|
|
symbol: str,
|
|
prev_close_raw: float,
|
|
cfg,
|
|
is_st: bool = False,
|
|
) -> bool:
|
|
"""涨停封板（一字板或 T 字板）→ 买单拒单。"""
|
|
board = get_board(symbol)
|
|
up = limit_up_price(prev_close_raw, limit_ratio(board, is_st), cfg.pricetick)
|
|
return is_one_word_lock(bar, up) or is_t_lock(bar, up)
|
|
|
|
|
|
def is_locked_for_sell_symbol(
|
|
bar: pd.Series,
|
|
symbol: str,
|
|
prev_close_raw: float,
|
|
cfg,
|
|
is_st: bool = False,
|
|
) -> bool:
|
|
"""跌停封板（一字板或跌停 T 字板）→ 卖单拒单。"""
|
|
board = get_board(symbol)
|
|
down = limit_down_price(prev_close_raw, limit_ratio(board, is_st), cfg.pricetick)
|
|
return is_one_word_lock(bar, down) or _is_t_lock_down(bar, down)
|