"""A 股涨跌停纯函数(板块表 + 封板判断)。 全函数无副作用,全部用 **raw** 价格(spec §3.3 / §6.2)。 封板判据: - 严格一字板:open=high=low=close=limit_price → 无对手盘 - T 字板(涨停):开=涨停 收=涨停 lowopen → 盘中反弹过,对称保守拒单 精度约定: - 涨跌停价用 Decimal ROUND_HALF_UP(非 banker's round),消除 5.55×1.10→6.10 类问题。 - 封板价格相等用 math.isclose 容差比较(pricetick/2), 消除 5.5600000000000005 == 5.56 漏判(review C1)。 """ import math from decimal import ROUND_HALF_UP, Decimal import pandas as pd from .models import AccountConfig # ---- 板块分类(按代码前缀)---- 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" # 沪/深主板 def lot_size_for(symbol: str) -> int: """最小交易手数对应股数:科创板 200(1 股递增),其余板块 100。""" return 200 if get_board(symbol) == "star" else 100 _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] # ---- 涨跌停价(ROUND_HALF_UP 四舍五入取整)---- def _round_half_up_to_tick(value: float, pricetick: float) -> float: """四舍五入到 pricetick 整数倍(用 Decimal 避免 banker's rounding)。""" d_value = Decimal(str(value)) / Decimal(str(pricetick)) d_quantized = d_value.quantize(Decimal("1"), rounding=ROUND_HALF_UP) return float(d_quantized * Decimal(str(pricetick))) def limit_up_price(prev_close_raw: float, ratio: float, pricetick: float) -> float: return _round_half_up_to_tick(prev_close_raw * (1 + ratio), pricetick) def limit_down_price(prev_close_raw: float, ratio: float, pricetick: float) -> float: return _round_half_up_to_tick(prev_close_raw * (1 - ratio), pricetick) # ---- 价格相等(容差比较,处理浮点尾数差)---- def _price_eq(a: float, b: float, pricetick: float = 0.01) -> bool: """浮点等价比较(容差 pricetick/2)。 例:5.5600000000000005 与 5.56 在 pricetick=0.01 下视为相等。 相对比较(<, >)保留原语义,不用此函数。 """ return math.isclose(a, b, abs_tol=pricetick / 2) # ---- 封板形态 ---- def is_one_word_lock(bar: pd.Series, limit_price: float, pricetick: float = 0.01) -> bool: """严格一字板:开=高=低=收=limit_price(容差比较)。""" return bool( _price_eq(bar["open"], limit_price, pricetick) and _price_eq(bar["high"], limit_price, pricetick) and _price_eq(bar["low"], limit_price, pricetick) and _price_eq(bar["close"], limit_price, pricetick) ) def is_t_lock(bar: pd.Series, limit_price: float, pricetick: float = 0.01) -> bool: """T 字板(涨停型):开=涨停、收=涨停、low bool: """T 字板(跌停型,对称):开=跌停、收=跌停、high>open(盘中反弹过跌停)。""" return bool( _price_eq(bar["open"], limit_price, pricetick) and _price_eq(bar["close"], limit_price, pricetick) and bar["high"] > bar["open"] ) # ---- 正式入口(带 symbol,自动判板块)---- def is_locked_for_buy_symbol( bar: pd.Series, symbol: str, prev_close_raw: float, cfg: AccountConfig, is_st: bool = False, listing_days: int = 0, ) -> bool: """涨停封板(一字板或 T 字板)→ 买单拒单。 新股前 5 日无涨跌幅限制(创业/科创/北交所):listing_days<5 → 不锁。 主板新股首日 ±44% 首版未实现(TODO),按普通 ±10% 处理。 """ board = get_board(symbol) if 1 <= listing_days <= 5 and board in ("gem", "star", "bse"): return False up = limit_up_price(prev_close_raw, limit_ratio(board, is_st), cfg.pricetick) return is_one_word_lock(bar, up, cfg.pricetick) or is_t_lock(bar, up, cfg.pricetick) def is_locked_for_sell_symbol( bar: pd.Series, symbol: str, prev_close_raw: float, cfg: AccountConfig, is_st: bool = False, listing_days: int = 0, ) -> bool: """跌停封板(一字板或跌停 T 字板)→ 卖单拒单。 新股前 5 日无涨跌幅限制(创业/科创/北交所):listing_days<5 → 不锁。 """ board = get_board(symbol) if 1 <= listing_days <= 5 and board in ("gem", "star", "bse"): return False down = limit_down_price(prev_close_raw, limit_ratio(board, is_st), cfg.pricetick) return is_one_word_lock(bar, down, cfg.pricetick) or _is_t_lock_down(bar, down, cfg.pricetick)