第 27 讲：A 股回测引擎架构：向量化 vs 事件驱动

本讲核心目标

在量化研究从实验室走向数亿真金白银实盘的过程中，**回测引擎（Backtesting Engine）**是检验策略生死的第一道防线。初级量化研究员常沉迷于在 Jupyter Notebook 中使用 Pandas/NumPy 数组进行点积求和的“向量化回测”，误以为这就是策略未来的真实净值；然而一旦上线实盘，由于无法处理资金不足挂单排队、T+1 仓位锁定、盘中流动性枯竭与分红送转除权等物理限制，策略净值瞬间发生灾难性漂移。

本讲将深入剖析向量化回测（Vectorized）与事件驱动回测（Event-Driven / DES）的本质差异；推导符合 A 股微观结构的时间推进状态机；手写工业级双账本对账系统（现金簿与持仓簿）；并彻底理顺信号生成、开盘估值、撮合成交、收盘盯市与下一日 T+1 解冻的全链路闭环！

1. 经典文献与工业界理论基石

构建高保真回测系统所需的经典量化文献与微观结构著作：

López de Prado, Marcos (2018). Advances in Financial Machine Learning. John Wiley & Sons.
第 11 章与第 12 章深入剖析了回测三大谎言（过拟合、幸存者偏差、未来函数），奠定了事件驱动仿真（Event-Driven Simulation）与合成压力测试的工业标准。
Harris, Larry (2003). Trading and Exchanges: Market Microstructure for Practitioners. Oxford University Press.
市场微观结构权威指南，详尽解析了订单类型、限价簿（LOB）撮合机制、交易时间切片与清算交收流程。
Farmer, J. Doyne, & Foley, Duncan (2009). "The economy needs agent-based modelling", Nature, 460(7256): 685-686.
阐释了基于智能体与离散事件模拟（DES）对于捕捉金融市场非线性反馈与流动性冲击的不可替代性。
Cont, Rama (2001). "Empirical properties of asset returns: stylized facts and statistical issues", Quantitative Finance, 1(2): 223-236.
总结了高频与日频金融收益率的肥尾、波动率聚集与自相关微观结构实证特征。

2. 两种回测架构的深度技术对比

在买方研发流水线中，向量化与事件驱动并非非此即彼，而是服务于不同研发阶段的双引擎：

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

评测维度	向量化回测（Vectorized Engine）	事件驱动回测（Event-Driven DES Engine）
底层数据结构	2D/3D 稠密矩阵（NumPy / Polars 向量化点乘）	离散事件队列（Time Queue）、订单簿与复式记账账本
计算时间复杂度	$O (T \times N)$ ，GPU/SIMD 并行加速，毫秒级完成数年回测	$O (\sum Events)$ ，逐事件推进状态，通常需数秒至数分钟
资金约束处理	❌ 无法动态跟踪现金；目标权重合计为 1 即假设资金无限充裕	✅ 严格受制于可用现金余额；现金不足时自动按优先级缩单或拒单
T+1 与冻结管理	❌ 无法区分当期买入与历史可卖底仓	✅ 精确跟踪持仓分仓日历，日终买入进入冻结仓，次日开盘解冻
流动性与挂单排队	❌ 假设无论挂单多大均能按收盘价 $100 %$ 成交	✅ 设定单笔成交不得超过开盘/全天成交量的 $10 % \sim 20 %$ ，超额部分部分成交或撤单
除权除息现金流	❌ 粗暴依赖复权价格，掩盖了分红到账日的现金流断流风险	✅ 登记日冻结红利、除权日确认应收资产、派息日现金解冻可用
代表开源项目	vectorbt, pyfolio, fastquant	Backtrader, Zipline-reloaded, Qlib, NautilusTrader

3. A 股单交易日全真离散时间推进状态机

为了杜绝任何隐性未来信息渗透，事件驱动引擎必须建立严格符合沪深交易所结算规则的时间推进状态机（State Machine）：

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

关键时点原则：

决策时点严格性：第 $t$ 日开盘的调仓决策，只能读取 $t - 1$ 日及之前的特征与行情；执行函数必须强制校验 signal_date < trade_date；
开盘估值与收盘估值完全隔离：开盘撮合只能使用 $P_{open}$ ，严禁在开盘撮合逻辑中偷看当天的 $P_{high}, P_{low}, P_{close}$ ；
先卖后买，资金闭环：在现金有限的前提下，组合调仓必须遵循“先撮合卖出单释放可用现金，再按可用资金规模搜索并撮合买入单”的铁律。

4. 现代量化账户恒等式与复式记账法

在生产级回测系统中，任何凭空产生的利润或隐性消失的费用都是灾难。账户在每一个微观时点 $t$ 必须严格满足资产负债恒等式：

E_{t} = C_{t} + D_{t}^{receivable} + \sum_{i = 1}^{N} Q_{i, t} P_{i, t}

其中：

$E_{t}$ ：投资组合总资产净值（Equity / NAV）；
$C_{t}$ ：当前可用现金余额（Available Cash），可立即用于买入新股票；
$D_{t}^{receivable}$ ：应收股利/待到账款项（Dividends Receivable），已通过股权登记日确认，计入基金总资产净值，但在红利发放日之前严禁作为买入现金使用；
$Q_{i, t}$ ：第 $i$ 只股票的当前总持仓股数（包含普通可卖股、今日买入冻结股与未上市送转股）；
$P_{i, t}$ ：当前盯市价格（开盘估值用 $P_{open}$ ，日终估值用 $P_{close}$ ）。

4.1 买入成交时的复式记账更新

当以价格 $P_{fill}$ 成功买入 $Q_{fill}$ 股股票时：

{\begin{cases} C_{t +} = C_{t -} - Q_{fill} P_{fill} - {Fee}_{buy} - {Impact}_{buy} \\ Q_{i, t +}^{total} = Q_{i, t -}^{total} + Q_{fill} \\ Q_{i, t +}^{sellable} = Q_{i, t -}^{sellable} (T+1 锁定，可卖股数绝不增加！) \end{cases}

4.2 卖出成交时的复式记账更新

当以价格 $P_{fill}$ 成功卖出 $Q_{fill}$ 股股票时：

{\begin{cases} C_{t +} = C_{t -} + Q_{fill} P_{fill} - {Fee}_{sell} - {Impact}_{sell} \\ Q_{i, t +}^{total} = Q_{i, t -}^{total} - Q_{fill} \\ Q_{i, t +}^{sellable} = Q_{i, t -}^{sellable} - Q_{fill} \end{cases}

5. 工业级事件驱动引擎实战

在 tutorial/examples/lecture_27_vectorized_vs_event_driven_backtest.py 中，我们构建了生产级 AshareEventDrivenEngine：

实现了 PositionLedger（持仓簿）与 CashLedger（现金簿）；
严格按交易日历推进，支持 T+1 动态解冻与零股（Lot Rule）处理；
开盘估值与收盘估值严格物理隔离，原生支持部分成交与资金不足缩量；
逐日输出可对账的资产负债快照序列。

可在终端直接运行验证：

bash

python3 -m examples.lecture_27_vectorized_vs_event_driven_backtest

"""日频账户参考实现：前日信号→保守开盘撮合→收盘盯市与登记日权益快照。"""
from dataclasses import dataclass
from enum import Enum
import numpy as np
import polars as pl
from examples.lecture_28_transaction_friction_and_market_impact import AshareFrictionEngine
from examples.lecture_29_matching_rules_limits_suspensions import (
    AshareExecutionMatchingSimulator, CorporateActionBook, DailyMarketBar, LotRule, as_date)


class OrderDirection(Enum):
    BUY = 'BUY'
    SELL = 'SELL'


class OrderStatus(Enum):
    FILLED = 'FILLED'
    PARTIAL = 'PARTIAL'
    REJECTED = 'REJECTED'


@dataclass
class Order:
    order_id: int
    trade_date: str
    symbol: str
    direction: OrderDirection
    requested_shares: int
    shares: int
    price: float
    total_friction: float
    status: OrderStatus
    reason: str


@dataclass
class Position:
    symbol: str
    total_shares: int = 0
    available_shares: int = 0
    unlisted_shares: int = 0
    last_price: float = 0.

    @property
    def market_value(self):
        return self.total_shares * self.last_price


class AshareEventDrivenBacktester:
    def __init__(self, initial_cash=10_000_000., *, trading_calendar,
                 friction_engine=None, lot_rules=None, corporate_actions=(), slippage_fixed=.01):
        if not np.isfinite(initial_cash) or initial_cash <= 0:
            raise ValueError('初始现金须有限且为正')
        calendar = [as_date(d) for d in trading_calendar]
        if not calendar or calendar != sorted(set(calendar)):
            raise ValueError('必须提供唯一、递增的经核验交易日历')
        self.calendar = calendar
        self.initial_cash = self.cash = float(initial_cash)
        self.dividend_receivable = 0.
        self.friction = friction_engine or AshareFrictionEngine()
        self.matcher = AshareExecutionMatchingSimulator(slippage_fixed)
        self.actions = CorporateActionBook(corporate_actions)
        self.lot_rules = lot_rules or {}
        self.positions, self.orders, self.daily_snapshots = {}, [], []
        self.current_date = None
        self.closed = False
        self.used_liquidity = {}
        self.daily_fees = 0.

    def _position(self, symbol):
        return self.positions.setdefault(symbol, Position(symbol))

    @property
    def total_equity(self):
        return self.cash + self.dividend_receivable + sum(p.market_value for p in self.positions.values())

    def _begin_day(self, day, bars):
        if day not in self.calendar or (self.current_date is not None and day < self.current_date):
            raise ValueError('非交易日或日期倒退')
        if self.current_date == day:
            if self.closed:
                raise ValueError('当日已收盘，不能再执行开盘订单')
            return
        if self.current_date is not None:
            if not self.closed or self.calendar.index(day) != self.calendar.index(self.current_date) + 1:
                raise ValueError('必须先逐日收盘记账，不能跳过交易日')
        required = {s for s, p in self.positions.items() if p.total_shares}
        required |= {a.symbol for a in self.actions.actions if as_date(a.ex_date) <= day and a.bonus_share_ratio
                     and self.actions.entitlements.get(a.action_id, 0) > 0
                     and (a.action_id, 'EX') not in self.actions.processed}
        if not required <= set(bars):
            raise ValueError('缺少持仓/送股证券的当日估值，停牌也须提供明确估值')
        settlement = self.actions.process_day(day)
        # 只在交易日推进时解冻，未上市送股始终锁定。
        for pos in self.positions.values():
            pos.available_shares = pos.total_shares - pos.unlisted_shares
        self.cash += settlement.cash_delta
        self.dividend_receivable += settlement.receivable_delta
        for symbol, qty in settlement.shares_delta.items():
            pos = self._position(symbol)
            pos.total_shares += qty
            pos.unlisted_shares += qty
        for symbol, qty in settlement.listed_delta.items():
            pos = self._position(symbol)
            pos.unlisted_shares -= qty
            pos.available_shares += qty
        self.current_date, self.closed = day, False
        self.used_liquidity, self.daily_fees = {}, 0.

    def execute_rebalance(self, target_weights, market_bars, trade_date, *, signal_date,
                          allow_new_buys=True):
        """开盘执行。目标来自更早日期；未知成交量为 0。买单缩量留足全部成本。

        日内再次调用不会解冻，也不会重复使用已消耗的开盘流动性。
        allow_new_buys 可由第 31 讲的数据质量/风险控制建议传入。
        """
        day = as_date(trade_date)
        if as_date(signal_date) >= day:
            raise ValueError('开盘信号必须在之前日期已知')
        if not all(np.isfinite(w) and w >= 0 for w in target_weights.values()) or sum(target_weights.values()) > 1 + 1e-10:
            raise ValueError('目标权重须非负有限且合计不超过 1')
        required = set(target_weights) | {s for s, p in self.positions.items() if p.total_shares}
        if not required <= set(market_bars):
            raise ValueError('缺少当日开盘行情；不能用旧价格伪造成交')
        if any(not isinstance(b, DailyMarketBar) or s != b.symbol for s, b in market_bars.items()):
            raise ValueError('需要证券代码匹配的 DailyMarketBar')
        self._begin_day(day, market_bars)
        for s, bar in market_bars.items():
            if s in self.positions:
                self.positions[s].last_price = bar.open
        equity = self.total_equity
        start = len(self.orders)
        # 股票缺省目标为零；资金不足只缩小买入，不卖出今日新买持仓。
        for symbol in sorted(set(target_weights) | set(self.positions)):
            pos = self._position(symbol)
            if not pos.total_shares:
                continue
            bar = market_bars[symbol]
            rule = self.lot_rules.get(symbol, LotRule())
            target = rule.floor_buy(equity * target_weights.get(symbol, 0) / bar.open)
            requested = max(0, pos.total_shares - target)
            if requested:
                available = min(requested, pos.available_shares)
                quantity = available if available == pos.total_shares else rule.floor_buy(available)
                self._trade(symbol, 'SELL', requested, quantity, bar, rule)
        for symbol, weight in sorted(target_weights.items()):
            bar = market_bars[symbol]
            pos = self._position(symbol)
            rule = self.lot_rules.get(symbol, LotRule())
            target = rule.floor_buy(equity * weight / bar.open)
            requested = rule.floor_buy(max(0, target - pos.total_shares))
            if requested:
                self._trade(symbol, 'BUY', requested, requested if allow_new_buys else 0, bar, rule,
                            blocked_reason='RISK_CONTROL' if not allow_new_buys else None)
        self._check_account()
        return self.orders[start:]

    def _trade(self, symbol, direction, requested, quantity, bar, rule, blocked_reason=None):
        if quantity == 0:
            ok, filled, price, reason = False, 0, 0., blocked_reason or 'T_PLUS_ONE_OR_LOT'
        else:
            ok, filled, price, reason = self.matcher.simulate_order_execution(
                symbol, direction, quantity, bar, lot_rule=rule,
                used_liquidity=self.used_liquidity.get((symbol, direction), 0))
        fee = 0.
        if ok and direction == 'BUY':
            # 二分搜索可负担的申报数量；成本单调包括保底、过户与非线性冲击。
            lo, hi, affordable = 0, (filled - rule.min_buy) // rule.step, 0
            while lo <= hi:
                mid = (lo + hi) // 2
                q = rule.min_buy + mid * rule.step
                cost = self.friction.calculate_trade_friction(symbol, direction, q * price, bar.adv_20d, bar.daily_volatility)
                if q * price + cost.total_friction <= self.cash:
                    affordable, lo = q, mid + 1
                else:
                    hi = mid - 1
            if affordable < filled:
                reason = 'CASH_SIZED' if affordable else 'INSUFFICIENT_CASH'
            filled = affordable
            ok = filled > 0
        if ok:
            fee = self.friction.calculate_trade_friction(symbol, direction, filled * price,
                                                         bar.adv_20d, bar.daily_volatility).total_friction
            pos = self._position(symbol)
            pos.last_price = bar.open
            if direction == 'BUY':
                self.cash -= filled * price + fee
                pos.total_shares += filled
            else:
                self.cash += filled * price - fee
                pos.total_shares -= filled
                pos.available_shares -= filled
            self.used_liquidity[(symbol, direction)] = self.used_liquidity.get((symbol, direction), 0) + filled
            self.daily_fees += fee
        status = OrderStatus.REJECTED if not ok else OrderStatus.FILLED if filled == requested else OrderStatus.PARTIAL
        self.orders.append(Order(len(self.orders) + 1, self.current_date.isoformat(), symbol,
            OrderDirection(direction), requested, filled, price, fee, status, reason))

    def _check_account(self):
        if not np.isfinite(self.total_equity) or self.cash < -1e-7 or self.dividend_receivable < -1e-7:
            raise RuntimeError('账户资金恒等式或现金非负约束被破坏')
        for p in self.positions.values():
            if not 0 <= p.available_shares <= p.total_shares - p.unlisted_shares or p.unlisted_shares < 0:
                raise RuntimeError('可卖/锁定股份与总持仓不一致')

    def close_day(self, trade_date, closing_prices):
        if self.current_date != as_date(trade_date) or self.closed:
            raise ValueError('需要先执行当天开盘阶段，且每个交易日只收盘一次')
        held = {s for s, p in self.positions.items() if p.total_shares}
        if not held <= set(closing_prices) or any(not np.isfinite(closing_prices[s]) or closing_prices[s] <= 0 for s in held):
            raise ValueError('每个持仓必须有有限、正数的当日收盘估值')
        for s in held:
            self.positions[s].last_price = closing_prices[s]
        self.actions.capture_record_date(self.current_date, {s: p.total_shares for s, p in self.positions.items()})
        self._check_account()
        previous = self.daily_snapshots[-1]['total_equity'] if self.daily_snapshots else self.initial_cash
        snapshot = dict(trade_date=self.current_date, total_equity=self.total_equity,
            cash=self.cash, dividend_receivable=self.dividend_receivable,
            market_value=sum(p.market_value for p in self.positions.values()),
            fees=self.daily_fees, net_return=self.total_equity / previous - 1,
            cumulative_return=self.total_equity / self.initial_cash - 1)
        self.daily_snapshots.append(snapshot)
        self.closed = True
        return snapshot

    def get_equity_curve_dataframe(self):
        return pl.DataFrame(self.daily_snapshots)


if __name__ == '__main__':
    calendar = ['2025-01-02', '2025-01-03']
    engine = AshareEventDrivenBacktester(initial_cash=100000, trading_calendar=calendar)
    for day, signal, price in [('2025-01-02', '2024-12-31', 10.), ('2025-01-03', '2025-01-02', 10.2)]:
        bar = DailyMarketBar('A', price, price, price, price, 11., 9.,
                             opening_buy_liquidity=100000, opening_sell_liquidity=100000)
        for order in engine.execute_rebalance({'A': 1.}, {'A': bar}, day, signal_date=signal):
            print(order)
        engine.close_day(day, {'A': price})
    print(engine.get_equity_curve_dataframe())
    print('账户包含第 28 讲成本和第 29 讲撮合；未模拟连续竞价队列或券商柜台。')

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
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240

6. 成熟开源事件驱动回测框架选型与实盘对比

框架名称	语言核心	架构设计特色	A 股本土化适配程度	生产实盘柜台支持	适用机构场景
Backtrader	Python	经典松耦合事件驱动，支持多数据流同步与自定义观察器（Analyzers）	中等（需自行封装 T+1 与印花税）	支持 Interactive Brokers、Oanda	中小型私募、量化教学与快速实盘原型
Zipline-reloaded	Python / Cython	Quantopian 遗产，分时/日频全真撮合，内置 Pipeline 因子对齐	需通过三方库（如 pyfolio）适配 A 股规则	弱（偏研究）	美股基本面多因子量化研究
Qlib	Python / C++	微软开源，专为 AI/ML 量化设计，支持向量化与事件驱动混合	极高（原生支持 A 股日频、复权与 Alpha 评估）	需自行对接柜台	头部中频多因子机器学习量化团队
NautilusTrader	Rust / Python	极致性能的异步多线程 DES，纳秒级时间戳，专为高频设计	较低（需自行开发 A 股规则与数据源）	原生支持主流加密与外盘 FIX 协议	高频 HFT、统计套利与低延迟做市商

7. 本讲小结与核心自测

核心考点	工业标准回答要点	常见工程/学术陷阱
向量化回测最大的致命缺陷是什么？	无法处理资金依赖、状态依赖与流动性截断，将无法买入或停牌的股票假设为无摩擦成交。	拿着没有扣税、没有 T+1 约束的向量化高夏普曲线直接忽悠募资，实盘上线即穿仓。
T+1 机制在事件驱动引擎中如何严密建模？	持仓拆分为 `total_shares` 与 `sellable_shares`；日内买入仅增加总股数，次日日初推进时再将普通持仓转入可卖股数。	误把当日买入的股票直接加入可卖数量，在同一天内完成 T+0 违规卖出回转。
如何防止开盘撮合逻辑产生未来数据泄漏？	严格实行开盘接口与收盘接口物理分离，撮合仅接收开盘报价，严禁传入当天的 High、Low、Close。	撮合函数直接读取当天 Bar 对象的全天均价或最低价来判断成交，造成严重的未来函数漏洞。

在明确了事件驱动引擎的骨架后，回测结果是否真实，决定性地取决于底层交易摩擦精算。

在下一讲中，我们将进入 第 28 讲：真实交易摩擦全真模拟，深度推导印花税、机构最低佣金与著名的平方根法则（Square-Root Law）市场冲击模型！

第 27 讲：A 股回测引擎架构：向量化 vs 事件驱动 ​

1. 经典文献与工业界理论基石 ​

2. 两种回测架构的深度技术对比 ​

3. A 股单交易日全真离散时间推进状态机 ​

关键时点原则： ​

4. 现代量化账户恒等式与复式记账法 ​

4.1 买入成交时的复式记账更新 ​

4.2 卖出成交时的复式记账更新 ​

5. 工业级事件驱动引擎实战 ​

6. 成熟开源事件驱动回测框架选型与实盘对比 ​

7. 本讲小结与核心自测 ​