claude_dev
f940841a9a
C1 CRITICAL: is_one_word_lock/is_t_lock 用 == 比较价格，浮点尾数差 (5.5600000000000005 vs 5.56) 导致一字板当可成交。改用 math.isclose 容差比较 (pricetick/2)。新增 _price_eq()，所有价格 == 改用之； 相对比较 (low<open, high>open) 保留原语义。 H1 HIGH: limit_up/down_price 用 Python round() 是 banker's rounding (round(610.5)=610)，导致 5.55×1.10→6.10 而非 6.11。改用 decimal.Decimal(str(x)) + ROUND_HALF_UP。 测试： - 5.05 一字涨停 (浮点尾数差 case) - 9.99×1.2 创业板一字板 - 0.35×1.05 ST T 字板 - 5.55×1.10 → 6.11 (banker's 消除) - 4.45×0.90 → 4.01 60 tests pass, 100% cov 保持。 Refs: Phase 3c C-S0 review findings [C1][H1]
123 lines
4.6 KiB
Python
123 lines
4.6 KiB
Python
"""A 股涨跌停纯函数（板块表 + 封板判断）。
|
||
|
||
全函数无副作用，全部用 **raw** 价格（spec §3.3 / §6.2）。
|
||
封板判据：
|
||
- 严格一字板：open=high=low=close=limit_price → 无对手盘
|
||
- T 字板（涨停）：开=涨停 收=涨停 low<open → 盘中砸过板，保守拒单
|
||
- T 字板（跌停）：开=跌停 收=跌停 high>open → 盘中反弹过，对称保守拒单
|
||
|
||
精度约定：
|
||
- 涨跌停价用 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
|
||
|
||
|
||
# ---- 板块分类（按代码前缀）----
|
||
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]
|
||
|
||
|
||
# ---- 涨跌停价（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<open（盘中砸过涨停）。"""
|
||
return bool(
|
||
_price_eq(bar["open"], limit_price, pricetick)
|
||
and _price_eq(bar["close"], limit_price, pricetick)
|
||
and bar["low"] < bar["open"]
|
||
)
|
||
|
||
|
||
def _is_t_lock_down(bar: pd.Series, limit_price: float, pricetick: float = 0.01) -> 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: "object",
|
||
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, 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: "object",
|
||
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, cfg.pricetick) or _is_t_lock_down(bar, down, cfg.pricetick)
|