Python量化交易系统：从行情数据到回测执行的可信工程

量化交易这件事，我在最初接触时最大的错觉就是：回测跑出一条漂亮的净值曲线，就等于策略成立了。后来才明白，从“能跑通”到“能信得过”，中间隔着的不是几十行代码，而是一整套关于数据、撮合、成本和执行的工程细节。这篇内容想聊的就是用 Python 搭一套完整的量化交易系统，从行情数据落库、回测引擎设计，一直走到模拟盘和策略执行，把中间那些文档里不会写、但真上手就一定会撞上的问题摊开来讲。适合已经会写 Python、也懂一点技术指标，但自己做出来的回测结果总是“看着很美、上手就崩”的读者。所有内容都是技术实现层面的讨论，涉及交易的部分一律在模拟环境里讲，不构成任何操作建议，真实资金相关的决定请自己判断。

1. 先分清“能跑通”和“能信得过”这两件事

我见过太多人（包括早期的我自己）拿到一份行情数据，写十几行 pandas，算个双均线交叉，画个收益曲线，看到最大回撤只有 8%、年化收益 25%，当时那个兴奋劲是真实存在的。然后把这套东西接到模拟盘上跑两个月，结果跟回测差出十万八千里。问题不在策略本身有多玄妙，而在于回测这套东西的“物理世界”建模得太粗糙了。

1.1 一条漂亮曲线背后通常藏着哪三笔账

回测失真，绝大多数情况下逃不出三个来源，我把它叫做三笔账。

第一笔是数据账。你用的价格序列如果是未经清洗的原始数据，里面可能混着停牌日、缺失日、未复权造成的价格跳空，甚至同一只标的因为代码变更而在时间轴上被劈成两段。你在这种数据上算出来的均线、波动率、相关性，全是错的，但错误又足够“平滑”，不会报错，所以你根本发现不了。

第二笔是撮合账。最典型的写法是“今天收盘价算出信号，就用今天收盘价成交”。这在代码层面只差一个 shift，但在现实里差的是整整一个交易日的信息优势——收盘价你其实要等到收盘后才拿得到，怎么可能用它成交。还有涨跌停一字板买不进、停牌当天不能交易、成交量小到你的单子能把盘口打穿，这些在回测里如果一律按“想成交就成交”处理，回测收益就会系统性偏高。

第三笔是成本账。佣金、税费、滑点，很多人只在最后减一个固定千分之几了事。真实的成本是跟你的换手率、下单方式、标的流动性强相关的。一个高频调仓的策略，如果成本建模偏乐观，回测和实盘的差距会大到让你怀疑人生。

这三笔账加起来的误差，足够把一个年化 20% 的策略打成负收益。所以我在做任何策略之前，会先花时间把回测引擎和数据的“可信度”调到自己能接受的水平，再谈策略逻辑。

1.2 自建引擎和现成框架怎么选

这个问题没有标准答案，但有个判断逻辑我一直在用：你对回测过程的每一环，是否需要在代码层面直接插手。

现成的回测框架好处是开箱即用、指标齐全、社区案例多，适合快速验证一个想法“有没有苗头”。但它的代价是抽象层很厚，你很难精确控制撮合细节和资金占用逻辑，一旦策略有比较特殊的仓位管理方式（比如分批建仓、条件单、跨品种对冲），就会不停地在框架的边界上打补丁，越打越别扭。

自建引擎正好相反：前期投入大，但每一行成交逻辑都是你自己写的，出问题的时候你知道该往哪儿看。我的做法是分阶段——想法验证阶段用轻量的自建向量化脚本快速筛，筛出有苗头的再搬进自建的事件驱动引擎里做严格回测。这样既不浪费时间在粗糙想法上做精细工程，也不会让一个“看起来不错”的东西直接进模拟盘。

下面这张表是我自己在选型时会对照的维度，纯经验值，不是绝对标准：

我的建议很直接：不要一上来就写大而全的引擎。先用一百行左右的脚本把想法筛一遍，确认逻辑站得住，再投入时间做工程化。反过来，如果你已经决定要认真做这件事，那引擎的撮合和成本模块一定要自己写，不要图省事。

2. 数据层是所有幻觉的源头，先把它洗干净

我把数据层放在整个系统最前面讲，不是因为它技术上最难，而是因为它的错误最隐蔽、影响最广。引擎写错了会报错，策略写错了会亏钱，但数据错了往往什么都不报，只是安静地让所有结论都偏一点点。

2.1 复权、停牌和时间戳对齐的实际处理

先说复权。同一只标的有不复权、前复权、后复权三种价格序列，它们的用途完全不同。做回测必须用后复权，原因是后复权的历史价格是固定的，不会因为今天发生了一次分红除权就把昨天的价格改掉；前复权虽然看着直观，但每次除权都会重算历史，你在上面算出来的历史信号会随时间漂移，这是个很恶心的隐性问题——同一份代码，今天跑和三个月后跑，回测结果不一样。

停牌的处理也有讲究。停牌日的价格是缺失的，如果你直接 dropna，时间轴上就会出现空洞，滚动窗口计算就会跨过停牌期去拿更早的价格，算出来的指标在语义上是错的。我的处理方式是保留交易日历，停牌日价格用前收盘价填充，同时打一个is_tradable标记，撮合的时候检查这个标记决定能不能成交。这样指标计算的连续性有了，成交的真实性也保住了。

时间戳对齐是另一个高频坑。行情数据里经常混着两种时间语义：一种是“这根 K 线代表的时间段”，一种是“这根 K 线生成的时间”。做日频问题不大，做分钟频就必须分清，否则很容易在 09:31 这根 K 线上把 09:30 到 09:31 的信息当成已知。我习惯在数据入库时就统一成“K 线结束时间”，并且在字段名上写清楚，比如bar_end_time，从命名上杜绝歧义。

下面是我实际在用的一个清洗函数骨架，去掉业务细节后的形态：

import pandas as pd def clean_bars(raw: pd.DataFrame, calendar: pd.DatetimeIndex) -> pd.DataFrame: df = raw.copy() df["trade_date"] = pd.to_datetime(df["trade_date"]) df = df.sort_values("trade_date") # 同一交易日多条记录只保留最后一条 df = df.drop_duplicates("trade_date", keep="last") # 对齐到交易日历，停牌日用前收盘填充 df = df.set_index("trade_date").reindex(calendar) df["is_tradable"] = df["close"].notna().astype(int) df["close"] = df["close"].ffill() df["open"] = df["open"].fillna(df["close"]) df["high"] = df["high"].fillna(df["close"]) df["low"] = df["low"].fillna(df["close"]) df["volume"] = df["volume"].fillna(0) # 兜底：极端异常值直接标记出来，不要静默过滤 df["price_jump"] = df["close"].pct_change().abs() > 0.25 return df.reset_index().rename(columns={"index": "trade_date"})

注意最后那个price_jump字段，我特意不自动删除异常价格，而是打标。原因是我踩过一次教训：某次清洗规则把一天的真实大涨误删了，导致后面所有信号都错位，而且因为删除得很“干净”，排查了整整一天才发现。现在我的原则是清洗只做标记和填充，删除动作必须人工确认。

2.2 生存偏差和未来函数，两个最难自查的陷阱

这两个问题是回测里最阴的，因为它们不会让代码报错，只会让结果变好看。

生存偏差说的是你的标的池只包含了“活到今天的”那些品种。如果你今天去拉一份当前可交易的列表，然后拿它去回测三年前，那三年前那些后来退市、停止交易的品种天然就被排除了，你的回测实际上是在“已知结局”的前提下选样本，收益当然漂亮。正确的做法是维护一份时点快照，每个交易日记录当天实际可交易的标的池，回测时按当天快照取池子。

未来函数的问题更常见。举个具体例子：df["close"].rolling(20).mean()在索引 i 位置算出来的是前 20 天（含当天）的均值，如果你在当天收盘价上直接用它产生信号并成交，就使用了当天的收盘价这个“未来信息”去决定当天成交，逻辑上是循环的。修正方式很简单，成交价用下一交易日的开盘价，信号照旧用今天收盘后计算。代码上就是一个shift(1)，但很多人第一次写的时候就是会漏。

还有一个更隐蔽的未来函数：行业分类和指数成分是用今天的版本。比如你回测五年前，用的是现在的行业归属表，那么当年发生过行业变更的标的就被错误归类了。这类问题在单标的策略里不明显，一旦做多标的轮动，误差会快速放大。我在做多标的策略前，会强制检查一遍所有“静态表”（行业表、成分表、上市状态表）是否带时间维度。

提示：写任何新策略前，先做一次“未来函数压力测试”——把回测的成交价整体往后挪一天，如果收益变化巨大，说明你的策略严重依赖未来信息，先别急着优化参数。

3. 回测引擎的骨架：事件驱动、撮合模拟与成本建模

引擎这一层，我自己的经验是：核心代码量其实不大，写清楚大概六七百行就够用，真正难的是把每一个决策点的边界条件想全。

3.1 为什么最终选了事件驱动而不是纯向量化

向量化回测写起来爽，一个 DataFrame 从头算到尾，几秒钟出结果。但它有个结构性缺陷：它无法表达“状态”。一旦你的策略涉及“已经持有多少仓位”“上一次止损是否触发”“今天还有多少可用资金”，向量化就变得极其别扭，你会被迫用一堆shift和累积变量去模拟原本很自然的状态机，代码可读性急剧下降，而且极易出错。

事件驱动的好处是把“时间推进”显式化了。每一根 K 线进来，引擎依次做三件事：更新账户状态、通知策略、处理产生的订单。这个过程和你真实交易时的心理流程是对应的，所以逻辑不容易写错。代价是慢，一次完整的日频回测可能要跑几十秒到几分钟，但这个时间花得值。

我的引擎主体结构大致是这样，去掉细节后：

from dataclasses import dataclass, field @dataclass class Bar: symbol: str dt: object open: float high: float low: float close: float volume: float is_tradable: int class Engine: def __init__(self, data_iter, strategy, broker, account): self.data_iter = data_iter # 按时间顺序产出当日全部 Bar self.strategy = strategy self.broker = broker self.account = account def run(self): for dt, bars in self.data_iter: # 1. 先用上一日收盘后的挂单做撮合 self.broker.match(dt, bars, self.account) # 2. 更新账户净值 self.account.mark_to_market(bars) # 3. 策略基于当日收盘数据产出新订单 orders = self.strategy.on_bars(dt, bars, self.account) # 4. 订单进入队列，等下一个交易日撮合 self.broker.submit(orders)

注意这个顺序里最关键的一点：撮合在策略之前，策略产出的订单不会当天成交。这一条从结构上就堵死了“用当天收盘价成交”的未来函数，比靠自觉写shift可靠得多。

3.2 撮合规则和成本建模的参数量怎么定

撮合这块，我会显式模拟四件事：成交价、可成交量、涨跌停限制、成本扣减。

成交价我一般用次日开盘价，遇到开盘一字板则判定无法成交。可成交量上，我设一个上限——单笔订单不超过该标的当日成交量的某个比例，超过就按比例部分成交。这个比例没有标准答案，日频策略我通常取 5% 到 10%，因为再高就明显脱离现实了。涨跌停的处理要小心，尤其在做小市值的时候，涨跌停当天的成交量并不小，但你不一定能买到，所以我的做法是涨停价买入、跌停价卖出直接判定失败并记录，事后统计这类失败订单的占比，如果占比超过 5%，说明策略选股落在流动性较差的区域，需要重新考虑。

成本建模我拆成三部分，用一张表说明各参数的取值逻辑：

滑点这块我特别想强调。很多人觉得滑点就是一个固定百分比，减掉就完了。但我实测下来，滑点跟订单规模的关系是超线性的——小额单子基本就是买卖价差的一半，大额单子会真的把盘口吃掉。所以我在引擎里做了个简单分段模型：订单金额占当日成交额比例低于某个阈值时用固定跳数，超过阈值后按比例递增。这个小改动让我的回测和模拟盘之间的偏差从 4 个百分点缩到 1 个百分点以内。

注意：成本参数的取值宁可保守也别乐观。回测里多扣一点成本，策略如果能活下来，实盘才有机会；回测里少扣，等于给自己发了一张假的成绩单。

另外还有一个容易被忽略的点：资金占用和结算规则。如果你的策略是 T+1 才能卖出的品种，回测里就必须严格限制当日买入的仓位当天不能平，否则高频进出会凭空多出很多次“完美交易”。这个约束我是在账户层做的，用一个frozen_position字典记录当日买入不可卖的数量，策略层调仓时先扣掉这部分。

4. 策略代码怎么写才像个工程而不是草稿

策略写多了会发现一个规律：亏钱的原因很少是逻辑不够聪明，多数是代码结构太乱导致根本没法验证。我早期写的策略，信号计算、仓位计算、下单逻辑全糊在一个函数里，改一个参数要动三处，回测出问题根本定位不了是哪一步错了。

4.1 把信号、目标仓位、订单拆成三层

现在我强制自己按三层写：第一层只负责算信号，输出是一个纯粹的数字或布尔值，不碰任何账户状态；第二层负责把信号翻译成目标仓位，考虑波动率、风险预算、仓位上限；第三层才根据目标仓位和当前仓位的差额生成订单。

拆开的好处是每一层都能单独测。信号层可以离线验证准确性，仓位层可以单独回放看仓位曲线是不是符合预期，订单层可以只数字段。这三层里任何一层出问题，范围都很明确。

class StrategyBase: def on_bars(self, dt, bars, account): signals = self.compute_signal(dt, bars) # 第一层：纯信号 targets = self.map_to_target(dt, signals, account) # 第二层：目标仓位 return self.build_orders(dt, targets, account) # 第三层：订单 def compute_signal(self, dt, bars): # 只读行情，不读账户，保证可独立复现 out = {} for sym, bar in bars.items(): ma_fast = self.fast_ma[sym].update(bar.close) ma_slow = self.slow_ma[sym].update(bar.close) if ma_fast is None or ma_slow is None: out[sym] = 0 else: out[sym] = 1 if ma_fast > ma_slow else 0 return out def map_to_target(self, dt, signals, account): # 波动率倒数加权，单标的仓位上限 20% vol = self.vol_estimator.snapshot() raw = {s: (1.0 / max(vol.get(s, 0.02), 1e-4)) * v for s, v in signals.items()} total = sum(raw.values()) if total <= 0: return {s: 0.0 for s in signals} return {s: min(w / total, 0.20) for s, w in raw.items()}

compute_signal里我用的是增量更新的移动平均，而不是每天重算整段序列。这个改动在日频上不明显，但当你跑几千个标的、十年数据的时候，性能差距是数量级的——重算版本可能跑半小时，增量版本只要一分钟。

4.2 参数配置与代码分离，让策略可以批量对照

第二件让我少走很多弯路的事，是把所有可调参数从代码里抽出来，放到独立的配置里。这样做最直接的好处是，同一份策略代码可以跑几十组参数，跑完直接做横向对比，不用改代码、不用怕改错。

CONFIG = { "fast_window": 10, "slow_window": 60, "vol_lookback": 20, "max_weight": 0.20, "rebalance_freq": 5, "stop_loss": 0.08, }

但配置化之后有个新问题会冒出来：参数一多，人就忍不住去优化。我以前干过一件事，拿了六组参数在一段历史上做网格搜索，挑了收益最高的那组。结果那组参数在样本外一塌糊涂。后来我才理解，参数优化本质上是在历史数据里找噪声，找到的“最优”很可能是过拟合。

现在我判断一组参数靠不靠谱，不看最高收益，看的是参数平原——把参数在邻域内做小幅扰动，如果收益变化剧烈（比如窗口从 60 改成 58，收益从 20% 掉到 5%），说明这组参数坐在悬崖边上，不可用；如果邻域内收益都差不多在一个区间内波动，说明逻辑本身稳定，参数只是次要因素。另外我会强制留出至少 30% 的数据作为样本外，从头到尾不参与任何参数选择，只在最后做一次验证。walk-forward 的方式我也用，但更耗时，一般用在准备上模拟盘的策略上。

5. 从回测到执行，那段真正容易翻车的路

回测跑出满意结果之后，很多人以为最难的部分过去了。恰恰相反，从回测到真实执行这一段，问题密度是最高的，因为这一步开始跟外部系统打交道，而外部系统不会按你的假设运行。

5.1 订单状态机和幂等，是执行层的生命线

订单不是“发出去就完成”的一次性动作。真实场景下，一笔订单会经历提交、已受理、部分成交、全部成交、已撤销、被拒绝这几种状态，而在网络抖动、接口超时的情况下，你甚至不知道自己发出去的那笔单子到底成功了没有。

如果没有状态机，代码会变成这样：发单 → 等一会儿 → 查一下 → 有就记账。这套逻辑在顺利时能跑，一旦遇到超时就会出问题——你不知道该重发还是该查询，重发可能导致重复下单，不重发可能漏单。

我的做法是每笔订单带一个自己生成的唯一标识（客户端订单号），所有后续操作都以这个标识为准。重试的时候带着同一个标识发出去，由对方做去重；查询的时候也用这个标识去查，保证查到的就是自己那笔。这个机制在工程上叫幂等，听着高级，实现起来其实就是“不要依赖对方的返回值来判断，依赖自己带的唯一键”。

from enum import Enum class OrderState(Enum): PENDING = "pending" SUBMITTED = "submitted" PARTIAL = "partial" FILLED = "filled" CANCELLED = "cancelled" REJECTED = "rejected" class Order: def __init__(self, client_id, symbol, side, qty, price=None): self.client_id = client_id self.symbol = symbol self.side = side self.qty = qty self.price = price self.state = OrderState.PENDING self.filled_qty = 0 def can_retry(self): # 只有仍在待提交状态的订单允许重试，已提交的只能查询 return self.state == OrderState.PENDING

can_retry这个方法是我被坑出来的。有一次网络波动，我的重试逻辑把一批已经提交成功的订单又发了一遍，幸好当时是模拟环境，否则就是双倍仓位。现在我的重试逻辑第一步永远是查状态，只有确认订单从未被受理才允许重发。

5.2 模拟盘和实盘之间，那些必须逐条核对的落差

模拟盘跑通了不代表实盘能跑通，这两者之间的落差清单我列一下，都是实际遇到过的：

• 成交价差异：模拟盘多数按盘口价或最新价成交，实际成交可能因为排队靠后而略差，这个差异会体现在小单高频策略上格外明显。
• 可用资金计算口径：有的系统把冻结资金算进可用，有的不算，导致下单时资金校验通过但实际被拒。
• 交易日历不一致：你的本地日历和交易端日历在节假日、临时调整上可能不同步，导致某天策略想交易但市场没开。
• 数据延迟：你本地的行情落库时间和交易端能看到的行情时间不完全一致，做分钟级策略时要特别留意。
• 最小交易单位：股数必须是整手，回测里按比例算出来的仓位需要取整，取整方向会影响实际仓位偏差。

我现在的做法是，任何策略在上真实资金之前，必须在模拟环境里连续跑通至少四周，并且每天做一次对账：把策略自己记录的持仓和成交，跟执行端导出的流水逐条比对，差异超过阈值就立刻停下来排查。这个对账脚本不到一百行，但它救过我至少两次。

6. 风控和运维：把“意外”当成必然来设计

我做这件事久了之后有个心态变化：不再追求“策略永远不出问题”，而是假设它一定会出问题，然后设计成“出问题的时候损失可控、能快速恢复”。这个思路上的转变，比任何策略优化都重要。

6.1 三道闸门：仓位、单日损失与异常波动

第一道是仓位闸门，控制的是单标的、单行业、总仓位三个上限。单标的上限我一般设在 20% 以内，行业上限会更保守，因为相关性高的标的会一起跌。总仓位上限则跟你对市场的判断有关，但这个值不该由单次信号决定，应该是配置里写死的硬约束。

第二道是单日损失闸门。我会预设一个当日亏损阈值，一旦触及，当天的所有开仓动作全部暂停，只允许平仓。这个机制的意义不在于止损本身，而在于防止情绪化操作——人在连续亏损的时候容易做出一连串糟糕决定，一个自动的刹车比意志力可靠得多。

第三道是异常波动闸门。当某个标的或者整体组合的波动率突然放大到历史均值的若干倍，说明市场结构在发生变化，此时策略依赖的假设可能已经不成立了。我会把这些标的暂时移入观察池，不新开仓，等波动回落到正常区间再恢复。

def pre_trade_check(order, account, limits, market_state): # 1. 仓位闸门 if account.weight(order.symbol) + order.notional / account.equity > limits.max_weight: return False, "single_weight_exceeded" if account.total_weight() + order.notional / account.equity > limits.max_total: return False, "total_weight_exceeded" # 2. 单日损失闸门 if account.daily_pnl / account.equity < -limits.daily_loss_limit: return False, "daily_loss_limit_hit" # 3. 异常波动闸门 if market_state.vol_ratio(order.symbol) > limits.vol_spike_threshold: return False, "volatility_spike" return True, "ok"

这个函数的返回值我特意做成了二元组，带上原因码，所有被拒绝的订单都会写日志。日积月累之后，这份日志本身就是很有价值的分析材料——你能看到哪些限制条件被触发的频率最高，从而判断自己的参数是不是设得太紧。

6.2 日志、对账和一个必须存在的总开关

日志这块，我的要求是“出问题时能只靠日志还原全过程”。所以关键事件全部结构化输出：订单提交、订单状态变更、成交回报、资金变动、风险拦截、异常堆栈。字段包括时间、标的、方向、数量、价格、原因码。写成 JSON 行格式，方便后续直接读进 DataFrame 做分析。

对账前面提过，再补一句细节：对账不只要比持仓数量，还要比成本价。因为成交价的小偏差会累积到成本价里，成本价错了，后面的止盈止损阈值就全错了，而且这种错误是渐进式的，很晚才会暴露。我一般设一个 0.5% 的成本价偏差告警线，超过就人工介入。

最后一个必须存在的东西是总开关。不管策略写得多稳妥，一定要有一个能在一秒内停掉全部自动交易动作的开关。实现起来可以简单到一个标志位文件，程序每次下单前检查这个文件存不存在，存在就直接不下单。再配合一个一键平仓的独立脚本，两者不要放在同一个进程里，避免进程挂掉的时候你连止损都做不了。

注意：总开关和平仓脚本要提前演练一遍。我见过有人写了开关但从来没测过，真出事的时候发现权限不对、路径不对、参数不对，白白耽误时间。

7. 一些我在长期实践里攒下的小结论

前面讲的都是结构层面的东西，最后补充几个更细碎的经验，属于那种“知道了能少走点弯路”的类型。

回测的初始资金不要设得太理想化。有些人用一千万的初始资金去跑一个消费类标的的策略，回测里随心所欲地满仓进出一只流动性一般的标的，实盘里这个资金量根本进不去。我的习惯是按照自己真实能投入的量级来设，这会让回测结果更接近现实，也避免了策略“纸上很强、实盘放不下”的尴尬。

换手率一定要在回测报告里显式列出来。很多人只看收益和回撤，忽略了年换手率这个指标。一个年换手 20 倍的策略，就算成本建模保守，实盘也会因为各种摩擦而打折严重。我的经验是日频策略年换手控制在个位数比较健康，超过 10 倍就要认真审视每一笔调仓的必要性。

策略的“空仓能力”值得单独设计。很多策略默认满仓运行，市场环境变了也不减仓，结果就是回撤全吃。在仓位层加入一个基于市场状态的总仓位调节因子，虽然会牺牲一部分上涨时的收益，但回撤控制的效果往往比优化选股更明显。

还有一点是关于心态的。我自己最难受的阶段不是策略亏钱，而是策略亏钱但我不知道原因——分不清是策略逻辑失效了，还是数据出问题了，还是执行环节漏了单。后来我把大量时间花在“可观测性”上：更细的日志、更全的对账、更直观的日报，这些工作不产生任何收益，但它让我在出问题的时候能在十分钟内定位，而不是花两天。现在我给自己定的规矩是，新功能上线前，先想清楚“这个功能如果出错了，我怎么发现”，答不上来就先别上线。