diff --git a/sanguo_trader/limit.py b/sanguo_trader/limit.py new file mode 100644 index 0000000..2f73c21 --- /dev/null +++ b/sanguo_trader/limit.py @@ -0,0 +1,94 @@ +"""A 股涨跌停纯函数(板块表 + 封板判断)。 + +全函数无副作用,全部用 **raw** 价格(spec §3.3 / §6.2)。 +封板判据: +- 严格一字板:open=high=low=close=limit_price → 无对手盘 +- T 字板(涨停):开=涨停 收=涨停 lowopen → 盘中反弹过,对称保守拒单 +""" +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 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) diff --git a/tests/trader/test_limit.py b/tests/trader/test_limit.py new file mode 100644 index 0000000..18bdcf0 --- /dev/null +++ b/tests/trader/test_limit.py @@ -0,0 +1,153 @@ +"""A 股涨跌停纯函数测试(板块表 + 封板判断)。全用 raw 价格。""" +import pandas as pd + +from sanguo_trader.limit import ( + get_board, + limit_ratio, + limit_up_price, + limit_down_price, + is_one_word_lock, + is_t_lock, + is_locked_for_buy_symbol, + is_locked_for_sell_symbol, +) +from sanguo_trader.models import AccountConfig + +CFG = AccountConfig(initial_capital=1_000_000) + + +def bar(open, high, low, close): + return pd.Series({"open": open, "high": high, "low": low, "close": close}) + + +# ---- 板块分类 ---- +def test_board_classification(): + assert get_board("600000") == "main" # 沪市主板 + assert get_board("000001") == "main" # 深市主板 + assert get_board("300750") == "gem" # 创业板 + assert get_board("301128") == "gem" # 创业板(301) + assert get_board("688981") == "star" # 科创板 + assert get_board("689009") == "star" # 科创板(689) + assert get_board("830799") == "bse" # 北交所 + assert get_board("920002") == "bse" # 北交所(920) + + +# ---- 幅度查表 ---- +def test_limit_ratio(): + assert limit_ratio("main", is_st=False) == 0.10 + assert limit_ratio("gem", is_st=False) == 0.20 + assert limit_ratio("star", is_st=False) == 0.20 + assert limit_ratio("bse", is_st=False) == 0.30 + # ST 统一 5% + assert limit_ratio("main", is_st=True) == 0.05 + assert limit_ratio("gem", is_st=True) == 0.05 + assert limit_ratio("star", is_st=True) == 0.05 + assert limit_ratio("bse", is_st=True) == 0.30 # 北交所 ST 同 30% + + +# ---- 涨跌停价(按 pricetick 取整)---- +def test_limit_up_price_rounds_to_pricetick(): + # 10.00 * 1.10 = 11.00 + assert limit_up_price(10.0, 0.10, 0.01) == 11.0 + # 9.99 * 1.20 = 11.988 → 11.99 + assert limit_up_price(9.99, 0.20, 0.01) == 11.99 + # 5.00 * 1.30 = 6.5 + assert limit_up_price(5.0, 0.30, 0.01) == 6.5 + + +def test_limit_down_price_rounds_to_pricetick(): + # 10.00 * 0.90 = 9.00 + assert limit_down_price(10.0, 0.10, 0.01) == 9.0 + # 10.00 * 0.80 = 8.00 + assert limit_down_price(10.0, 0.20, 0.01) == 8.0 + + +# ---- 一字板 / T 字板 ---- +def test_one_word_lock_detected(): + up = limit_up_price(10.0, 0.10, 0.01) + assert is_one_word_lock(bar(11.0, 11.0, 11.0, 11.0), up) is True + # 非一字 + assert is_one_word_lock(bar(11.0, 11.5, 10.8, 11.0), up) is False + assert is_one_word_lock(bar(11.0, 11.0, 10.5, 11.0), up) is False # T 字板非一字 + + +def test_t_lock_detected(): + up = limit_up_price(10.0, 0.10, 0.01) + # T 字板:开=涨停 收=涨停 lowopen(盘中反弹过) + assert is_locked_for_sell_symbol( + bar(9.0, 9.5, 9.0, 9.0), "600000", 10.0, CFG, is_st=False + ) is True + + +def test_sell_not_locked_when_opened(): + # 非跌停形态 → 可卖 + assert is_locked_for_sell_symbol( + bar(9.5, 9.8, 9.2, 9.6), "600000", 10.0, CFG, is_st=False + ) is False