第 29 讲：A 股特殊撮合规则代码级实现

本讲核心目标

许多从美股或加密货币转向 A 股的量化团队，往往会遭遇当头棒喝：A 股不是一个简单的连续撮合自由市场，而是充斥着涨跌停板限制、一字板被动锁仓、申报价格笼子、长期无故停牌以及极其繁琐的分红除权交收日历。

若回测系统对这些本土制度缺乏代码级保真模拟，不仅会导致回测产生“涨停板天天追进暴赚”、“跌停板瞬间清仓逃生”的虚假神话，更会导致实盘发生严重的资金交收透支违约。本讲将深入解构 A 股现行交易所撮合机制；手写涨跌停保守拒单与开盘流动性截断状态机；详解全面注册制 2% 申报价格笼子算法；并彻底打通股权登记日、除权除息日、派息到账日与送转股上市日的四阶段记账闭环！

1. 监管法规与交易所微观撮合制度

构建高保真撮合系统必须严格遵循沪深北交易所的核心规章：

上海/深圳证券交易所（2023）：《上海证券交易所交易规则》《深圳证券交易所交易规则》（全面注册制修订版）。
规定了集合竞价最大成交量原则、主板/双创板涨跌停幅度、全面注册制 2% 申报价格笼子与异常波动停牌停牌机制。
中国证券登记结算有限责任公司（ChinaClear, 2022）：《证券登记结算业务指南》与《权益分派业务操作指引》。
确立了股权登记日（Record Date）、除权除息日（Ex-Date）、资金交收日（Payout Date）与红利税差额补缴的标准化清算流程。
中国证券投资基金业协会（AMAC, 2017）：《关于发布中基协（AMAC）基金行业股票估值指数的通知》。
规范了公募与私募基金在面对个股长期停牌时，必须采用“AMAC 行业指数收益法”对停牌标的进行公允盯市估值的强制性准则。

2. A 股多层次涨跌停板与板上撮合物理学

2.1 多层次板块涨跌幅限制矩阵

不同板块在不同生命周期与风险状态下，适用的涨跌幅限制存在严格差异：

证券板块	证券代码前缀	正常交易日涨跌幅	ST / *ST 风险警示涨跌幅	IPO 上市初期规则
沪深主板	600, 601, 603, 000, 001	$\pm 10 %$	$\pm 5 %$	前 5 个交易日不设涨跌幅限制，随后 $\pm 10 %$
创业板（ChiNext）	300, 301	$\pm 20 %$	$\pm 20 %$	前 5 个交易日不设涨跌幅限制，随后 $\pm 20 %$
科创板（STAR Market）	688	$\pm 20 %$	$\pm 20 %$	前 5 个交易日不设涨跌幅限制，随后 $\pm 20 %$
北交所（BSE）	8xx, 43x, 920	$\pm 30 %$	$\pm 30 %$	首日无涨跌幅，随后 $\pm 30 %$

2.2 涨跌停撮合物理机制与“被动锁仓（Passive Lock-in）”

可放大查看细节，点击百分比恢复原始大小。

在量化回测中，必须采取买方保守拒单假设（Conservative Rejection Assumption）：

买单撞涨停：当股票开盘价等于涨停价时，由于排队时间优先原则，量化策略的外部买单极大概率无法排进成交，必须严格拒单，持仓与现金保持不变；
卖单撞跌停：当股票开盘价等于跌停价时，由于流动性彻底枯竭，策略卖单根本无法撮合出局，发生被动锁仓（Passive Lock-in）！该股票持仓必须被迫继续保留，并完全承受随后的下跌浮亏！

3. 全面注册制 2% 申报价格笼子机制（Order Price Collar）

自 2023 年全面注册制落地以来，沪深交易所对连续竞价阶段的所有限价申报施加了动态 2% 价格笼子限制，旨在彻底消灭“秒挂涨停价抢筹”或“秒挂跌停价砸盘”的算法行为：

3.1 申报价格笼子数学公式

设连续竞价当前盘口基准价格为 $P_{ref}$ （通常取对手方最优价格；若无对手方，取本方最优价格）：

买入限价申报上限： $P_{buy_order} \leq max (P_{ref} \times 1.02, P_{ref} + 10 ticks)$
卖出限价申报下限： $P_{sell_order} \geq min (P_{ref} \times 0.98, P_{ref} - 10 ticks)$

超过上述阈值的申报将被交易所撮合主机直接判定为废单拒单或进入暂存暂不受理状态。算法执行（如 VWAP/TWAP）必须实时跟踪盘口微步微调报单价格。

4. 停牌标的盯市估值：中基协（AMAC）行业指数法

当股票因重组、暴雷或市场异常被长期停牌（如 2015 年千股停牌潮，乐视网、康得新停牌数月至数年）时，若简单按“停牌前最后收盘价”估值，会导致基金净值严重虚高；一旦遭遇大额赎回，先赎回的投资者按虚高净值套现，后留下的投资者将承担巨额穿仓损失。

4.1 中基协行业指数收益法（AMAC Valuation）

公募与规范私募机构在股票停牌期间，按中基协发布的对应行业分类指数的相对涨跌幅，对停牌股票价格进行动态盯市重估：

P_{i, t}^{adjusted} = P_{i, suspend} \times \frac{I_{industry, t}}{I_{industry, t_{0}}}

其中：

$P_{i, suspend}$ ：停牌前最后一个交易日的收盘价；
$I_{industry, t_{0}}$ ：停牌当日该股票所属申万/中基协行业指数的收盘点位；
$I_{industry, t}$ ：当前估值日该行业指数的收盘点位。

若停牌期间该板块大跌 $30 %$ ，则该股票估值按 $70 %$ 计提公允价值，从而真实反映资产组合的实际风控净值。

5. 分红送转权益清算四阶段状态机（Corporate Actions）

分红送转绝不是简单的“在除权日改一下价格”，它横跨四个不同的法定交易日历节点，必须通过严密的对账状态机进行处理：

可放大查看细节，点击百分比恢复原始大小。

5.1 除权除息理论参考价方程

在除权除息日开盘前，交易所对前收盘价执行除权除息机械修正（去除现金红利与送转股膨胀）：

P_{ex} = \frac{P_{before} - d + p_{allot} \cdot r_{allot}}{1 + b + r_{allot}}

其中：

$P_{before}$ ：股权登记日收盘价；
$d$ ：每股派发税前现金红利（元）；
$b$ ：每股送红股与转增股本比例（如 10 送转 5 则 $b = 0.5$ ）；
$p_{allot}$ ：配股价格； $r_{allot}$ ：配股比例。

5.2 账户资产守恒性验证示例

假设登记日持有 100 股，登记日收盘价 10 元，总市值 1,000 元。公司公布“10 派 10 元转增 1 股”（即每股红利 $d = 1.0$ 元，转增比例 $b = 0.1$ ）：

除权参考价： $P_{ex} = \frac{10 - 1.0}{1 + 0.1} = \frac{9.0}{1.1} \approx 8.1818 元$
权益确认：
- 确认现金红利应收： $100 \times 1.0 = 100$ 元（ $D^{receivable}$ ，尚不可买入股票）；
- 确认总持股数量增加： $100 \times (1 + 0.1) = 110$ 股（其中 10 股为未上市锁定股）；
除权日开盘资产估值： $Total Assets = 110 \times \frac{9.0}{1.1} + 100 = 900 + 100 = 1, 000 元 (绝对守恒！)$

若回测系统在除权日直接将 100 元红利加进可用现金去买股票，或允许当天将 110 股全部卖掉，就会产生严重的资金透支与超额虚假卖空违规！

6. 工业级特殊撮合与公司行为账本实战

在 tutorial/examples/lecture_29_matching_rules_limits_suspensions.py 中，我们实现了生产级 AshareSpecialMatchingEngine：

实现严格的开盘涨跌停判定、停牌识别与保守拒单；
实现了单笔调仓基于开盘流动性上限的等比例部分成交（Partial Fill）；
实现了 CorporateActionTracker，完美处理股权登记日快照、应收股利冻结、派息日现金解冻与送转股上市的全流程对账。

可在终端直接运行验证：

bash

python3 -m examples.lecture_29_matching_rules_limits_suspensions

"""保守开盘撮合与登记日权益账本。日内排队、税补扣和零碎股分配不在本例范围。"""
from dataclasses import dataclass, field
from datetime import date
import numpy as np


def as_date(value):
    return value if type(value) is date else date.fromisoformat(str(value))


@dataclass(frozen=True)
class LotRule:
    min_buy: int = 100
    step: int = 100

    def __post_init__(self):
        if not isinstance(self.min_buy, int) or not isinstance(self.step, int) or min(self.min_buy, self.step) < 1:
            raise ValueError('最小申报量和步长须为正整数')

    def floor_buy(self, shares):
        return 0 if shares < self.min_buy else self.min_buy + int((shares - self.min_buy) // self.step) * self.step


@dataclass(frozen=True)
class DailyMarketBar:
    """撮合仅访问 open/限价/停牌及显式开盘可成交量；high/low/close 用于收盘展示。"""
    symbol: str
    open: float
    high: float
    low: float
    close: float
    limit_up_price: float | None
    limit_down_price: float | None
    is_suspended: bool = False
    opening_buy_liquidity: int = 0
    opening_sell_liquidity: int = 0
    adv_20d: float = 1e8       # 决策前已知的历史成交额
    daily_volatility: float = .02

    def __post_init__(self):
        if not np.isfinite([self.open, self.adv_20d, self.daily_volatility]).all() or min(self.open, self.adv_20d) <= 0 or self.daily_volatility < 0:
            raise ValueError('开盘价、ADV 须为正，波动率非负且有限')
        for q in (self.opening_buy_liquidity, self.opening_sell_liquidity):
            if not isinstance(q, int) or q < 0:
                raise ValueError('开盘可成交量须为非负整数，未知时为 0')
        for bound in (self.limit_up_price, self.limit_down_price):
            if bound is not None and (not np.isfinite(bound) or bound <= 0):
                raise ValueError('涨跌停价须为正；无限价时显式传 None')
        if self.limit_up_price is not None and self.open > self.limit_up_price + 1e-8:
            raise ValueError('开盘价超过涨停价')
        if self.limit_down_price is not None and self.open < self.limit_down_price - 1e-8:
            raise ValueError('开盘价低于跌停价')


class AshareExecutionMatchingSimulator:
    """不掌管持仓或现金。开盘触及不利限价时保守拒单，不推断盘中是否开板。"""
    def __init__(self, price_slippage=.01):
        if not np.isfinite(price_slippage) or price_slippage < 0:
            raise ValueError('固定滑点须非负且有限')
        self.slippage = price_slippage

    def simulate_order_execution(self, symbol, direction, order_shares, market_bar,
                                 *, lot_rule=None, used_liquidity=0):
        rule = lot_rule or LotRule()
        if symbol != market_bar.symbol or direction not in ('BUY', 'SELL'):
            raise ValueError('证券或交易方向不匹配')
        if not isinstance(order_shares, int) or order_shares <= 0 or used_liquidity < 0:
            raise ValueError('委托数量须为正整数')
        if market_bar.is_suspended:
            return False, 0, 0., 'SUSPENDED'
        upper, lower = market_bar.limit_up_price, market_bar.limit_down_price
        if direction == 'BUY':
            if upper is not None and market_bar.open >= upper - 1e-8:
                return False, 0, 0., 'OPEN_AT_LIMIT_UP'
            quantity = rule.floor_buy(min(order_shares, max(0, market_bar.opening_buy_liquidity - used_liquidity)))
            price = market_bar.open + self.slippage
            if upper is not None and price > upper:
                return False, 0, 0., 'SLIPPAGE_OUTSIDE_LIMIT'
        else:
            if lower is not None and market_bar.open <= lower + 1e-8:
                return False, 0, 0., 'OPEN_AT_LIMIT_DOWN'
            quantity = min(order_shares, max(0, market_bar.opening_sell_liquidity - used_liquidity))
            price = market_bar.open - self.slippage
            if price <= 0 or (lower is not None and price < lower):
                return False, 0, 0., 'SLIPPAGE_OUTSIDE_LIMIT'
        if quantity == 0:
            return False, 0, 0., 'NO_OPENING_LIQUIDITY_OR_LOT'
        return True, quantity, price, 'FILLED' if quantity == order_shares else 'PARTIAL'


@dataclass(frozen=True)
class CorporateAction:
    action_id: str
    symbol: str
    record_date: str
    ex_date: str
    cash_payment_date: str | None = None
    share_listing_date: str | None = None
    cash_div_per_share: float = 0.
    bonus_share_ratio: float = 0.

    def __post_init__(self):
        if not self.action_id or not self.symbol or as_date(self.record_date) >= as_date(self.ex_date):
            raise ValueError('事件须有唯一 ID，且登记日早于除权日')
        if not np.isfinite([self.cash_div_per_share, self.bonus_share_ratio]).all() or min(self.cash_div_per_share, self.bonus_share_ratio) < 0:
            raise ValueError('现金和股份权益须非负且有限')
        for amount, when in ((self.cash_div_per_share, self.cash_payment_date), (self.bonus_share_ratio, self.share_listing_date)):
            if amount and (when is None or as_date(when) < as_date(self.ex_date)):
                raise ValueError('必须提供不早于除权日的实际到账/上市日期')


@dataclass
class CorporateSettlement:
    cash_delta: float = 0.
    receivable_delta: float = 0.
    shares_delta: dict = field(default_factory=dict)
    listed_delta: dict = field(default_factory=dict)


class CorporateActionBook:
    """登记日收盘冻结数量；除权日确认应收/股份；支付与上市分开，重复调用无重复到账。"""
    def __init__(self, actions=()):
        self.actions = list(actions)
        if len({a.action_id for a in self.actions}) != len(self.actions):
            raise ValueError('公司行为 ID 重复')
        self.entitlements = {}
        self.processed = set()
        self.last_processed_date = None

    def capture_record_date(self, current_date, holdings):
        day = as_date(current_date)
        for a in self.actions:
            if as_date(a.record_date) == day:
                q = holdings.get(a.symbol, 0)
                if not isinstance(q, int) or q < 0:
                    raise ValueError('登记日持股数须为非负整数')
                if a.action_id in self.entitlements and self.entitlements[a.action_id] != q:
                    raise ValueError('不能覆盖已冻结的登记日权益')
                self.entitlements[a.action_id] = q

    def process_day(self, current_date):
        day = as_date(current_date)
        if self.last_processed_date is not None and day < self.last_processed_date:
            raise ValueError('公司行为账本不允许时间倒退')
        due = [a for a in self.actions if as_date(a.ex_date) <= day]
        # 先验证全部事件，避免异常造成半次结算。
        for a in due:
            if a.action_id not in self.entitlements:
                raise ValueError(f'缺少登记日持仓快照: {a.action_id}')
            q = self.entitlements[a.action_id] * a.bonus_share_ratio
            if not np.isclose(q, round(q), atol=1e-8, rtol=0):
                raise ValueError('零碎股需上游提供登记结算后的分配结果，本例不自动截断')
        result = CorporateSettlement()
        for a in due:
            quantity = self.entitlements[a.action_id]
            cash, shares = quantity * a.cash_div_per_share, round(quantity * a.bonus_share_ratio)
            for phase, when in [('EX', a.ex_date), ('PAY', a.cash_payment_date), ('LIST', a.share_listing_date)]:
                key = (a.action_id, phase)
                if when is None or as_date(when) > day or key in self.processed:
                    continue
                if phase == 'EX':
                    result.receivable_delta += cash
                    result.shares_delta[a.symbol] = result.shares_delta.get(a.symbol, 0) + shares
                elif phase == 'PAY':
                    result.cash_delta += cash
                    result.receivable_delta -= cash
                else:
                    result.listed_delta[a.symbol] = result.listed_delta.get(a.symbol, 0) + shares
                self.processed.add(key)
        self.last_processed_date = day
        return result


if __name__ == '__main__':
    matcher = AshareExecutionMatchingSimulator()
    for low in (11., 10.5):
        bar = DailyMarketBar('A', 11, 11, low, 10.8, 11, 9, opening_buy_liquidity=10000)
        print('后续最低价', low, matcher.simulate_order_execution('A', 'BUY', 100, bar))
    action = CorporateAction('A-DIV', 'A', '2025-01-02', '2025-01-03',
                             '2025-01-06', '2025-01-07', 1., .2)
    book = CorporateActionBook([action])
    book.capture_record_date('2025-01-02', {'A': 100})
    for day in ['2025-01-03', '2025-01-06', '2025-01-06', '2025-01-07']:
        print(day, book.process_day(day))

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184

7. 第七模块核心交付 Checklist

学完本模块，你的量化策略已经成功经受住了 A 股市场最严苛的“实盘微观物理检验”：

[x] 回测引擎架构辨析（第 27 讲）：
- 厘清了向量化回测与事件驱动回测的边界与职责分工；
- 掌握了符合 A 股制度的单日时间推进离散事件状态机（DES）与复式记账法。
[x] 交易摩擦全真模拟（第 28 讲）：
- 精确核算了印花税单边 $0.05 %$ 、机构佣金及最低 5 元保底陷阱；
- 推导了非线性平方根法则冲击模型，并解析求解了策略黄金 AUM 容量峰值闭式解。
[x] A 股特殊撮合规则代码级闭环（第 29 讲）：
- 实现了涨跌停板保守拒单、被动锁仓与 2% 申报价格笼子机制；
- 实现了停牌标的中基协行业指数估值调整法；
- 完整跑通了股权登记、除权除息、分红到账与送转上市的四阶段严密对账。

至此，我们的策略不仅在回测中跑出了真实可信的扣费净资产曲线，而且做好了上线实盘柜台的所有制度准备！

在接下来的 第八模块：生产级监控系统、业绩归因与实盘上线 中，我们将为策略装上全天候的业绩归因显微镜（Brinson 与 Barra 归因）与风控熔断预警看板（跟踪误差漂移与 PSI 特征监控），护航策略实盘长期稳健运行！

第 29 讲：A 股特殊撮合规则代码级实现 ​

1. 监管法规与交易所微观撮合制度 ​

2. A 股多层次涨跌停板与板上撮合物理学 ​

2.1 多层次板块涨跌幅限制矩阵 ​

2.2 涨跌停撮合物理机制与“被动锁仓（Passive Lock-in）” ​

3. 全面注册制 2% 申报价格笼子机制（Order Price Collar） ​

3.1 申报价格笼子数学公式 ​

4. 停牌标的盯市估值：中基协（AMAC）行业指数法 ​

4.1 中基协行业指数收益法（AMAC Valuation） ​

5. 分红送转权益清算四阶段状态机（Corporate Actions） ​

5.1 除权除息理论参考价方程 ​

5.2 账户资产守恒性验证示例 ​

6. 工业级特殊撮合与公司行为账本实战 ​

7. 第七模块核心交付 Checklist ​