Python算法交易全链路：回测撮合、实盘执行与滚动验证

简介：面向希望用 Python 搭建全自动交易系统的量化爱好者与进阶开发者，这份教程型代码包围绕行情数据获取、Pandas 数据处理、技术指标计算、策略回测、交易执行、风险与资金管理、实时运行及日志监控等环节展开，帮助读者理解从策略想法到自动化落地的完整链路。包内共 35 个文件，以 34 个 Python 脚本和 1 个 Markdown 说明为主，压缩包约 38KB，脚本覆盖收益风险指标、组合再平衡、突破策略、可视化、财务数据抓取与常用技术指标等实践片段，便于按主题检索和二次改造。已有 3962 人学习下载，适合具备一定 Python 与金融基础、需要参考代码框架和策略实现思路的读者。

1. 从一份 Python 算法交易源码说起：它能替你做什么

拆过几份标着「全自动」的算法交易代码之后，我越来越确信一件事：真正决定成败的不是策略本身有多花哨，而是回测和实盘有没有跑在同一套撮合逻辑上。这份 Python 量化资源值得逐行读的地方就在这里——它把行情接入、信号生成、仓位换算、订单执行串成了一条能跑通的闭环，你把定量交易想法塞进去，就能拿到一个带成本假设的净值曲线。适合已经会用 pandas 处理时间序列、想从「跑通一个回测」跨到「每天自动执行」的开发者；指望拿到一个挂上就能躺赚的黑匣子的人，大概率会失望。下面按数据、策略、撮合、实盘、验证的顺序往下拆，重点落在参数怎么取和最容易翻车的地方。

2. 拆开代码包：行情、信号、撮合、风控四层怎么分

拿到一份能实盘的算法交易系统，先别急着读策略文件。我一般的顺序是看目录结构，判断它有没有把「取数」「算信号」「撮合」「下单」拆成互相不认识的模块。拆开了，你换数据源只要改一个文件，换策略只要实现一个接口，出问题也能定位到具体层；没拆开，通常是一个几百行的 while 循环里混着 requests 和 if 判断，那种代码回测跑得再漂亮也不敢挂上去。

分层的价值不只是好看。回测和实盘最大的差异在于「成交假设」，如果取数和撮合缠在一起，你几乎不可能把同一份信号在两种成交模型下各跑一遍做对照。所以下面这四层，我在自己项目里是强制分开的，这份资源的组织方式也基本是这个思路。

2.1 数据层：把不同来源的 K 线对齐成一张表

数据层的目标很朴素：不管上游给的是接口返回值、CSV 还是数据库查询结果，最终都收敛成同一个结构——以交易日为索引，至少包含 open、high、low、close、volume 五列，价格必须是复权后的。这一步看着枯燥，但后面所有玄学问题一半出在这里。

统一 schema 的时候有三件事必须写死在代码里，而不是靠记忆：

• 时区与时间戳含义。日线的datetime到底代表当日开盘、收盘还是零点，不同数据源不一样。我习惯统一成「该交易日收盘时刻」，因为策略是在收盘后算信号、次日开盘执行。
• 复权口径。前复权适合回测（历史价格连续），后复权适合持仓成本核算。混用会让某次除权日的收益率凭空多出十几个点。
• 缺失与停牌。停牌日很多源是直接不给这一行，而不是给一行 volume=0。如果你用rolling(20)直接算均线，停牌会被无声跳过，等于用更长的自然日算了 20 个交易日，参数含义就变了。

常见做法是建一个交易日历，把行情reindex到日历上，停牌日价格用前收盘填充、volume 填 0，同时打一个is_tradable标记。撮合层看到不可交易就跳过下单，而不是让策略自己判断。

import pandas as pd def align_to_calendar(bars: pd.DataFrame, calendar: pd.DatetimeIndex) -> pd.DataFrame: """ bars: 原始行情, index 为交易日, 含 open/high/low/close/volume calendar: 完整交易日历 (含节假日剔除后的所有交易日) """ df = bars.reindex(calendar) # 缺失日补 NaN df["is_tradable"] = df["volume"].fillna(0) > 0 # 停牌日价格用前收盘填充, 避免 rolling 窗口被压缩 df["close"] = df["close"].ffill() for col in ("open", "high", "low"): df[col] = df[col].fillna(df["close"]) df["volume"] = df["volume"].fillna(0) return df

这段代码的关键在reindex加ffill的组合：reindex保证时间轴完整，ffill保证价格连续，is_tradable保证撮合层知道哪天不能动。参数上唯一需要你确认的是calendar的来源——必须和你的行情源用同一套节假日规则，否则会出现「日历说开市、行情没数据」的假缺口，这种缺口喂进均线会让信号抖动。

2.2 策略层：一个只输出目标仓位的纯函数

策略接口我强烈建议定义成「输入历史窗口，输出 0 到 1 之间的目标仓位比例」，而不是输出「买/卖」信号。原因很实际：买卖信号需要额外的状态机去维护当前持仓，一旦程序重启、订单部分成交、或者某天手动干预过账户，状态就对不上了；目标仓位是幂等的——今天算出 0.6，实盘发现当前只有 0.3，那就补到 0.6，明天再算出 0.6 就什么都不做。

纯函数还有两个好处：一是回测时可以随便并行、随便切片做滚动验证；二是单元测试好写，给定一段固定行情，输出必须完全一致，任何随机性都要显式传 seed。

接口大致长这样：generate_target(history: pd.DataFrame) -> float。history只包含当前时点之前的数据，策略内部不允许访问未来行。这条约束听起来是废话,但我在别人的代码里见过用整段 DataFrame 算close.mean()当阈值的写法,那就是标准的未来函数。

2.3 撮合层：回测循环里必须写死的四个约束

回测循环是整个包里最该逐行读的部分。一个诚实的最小实现必须包含四件事：信号延迟一天执行、双边手续费、按不利方向计的滑点、最小交易单位取整。

import pandas as pd def run_backtest(bars: pd.DataFrame, targets: pd.Series, init_cash=100_000, fee_rate=0.0003, slippage=0.0005, lot=100): """ bars: 含 open/close/is_tradable 的行情 targets: 策略在 t 日收盘后产生的目标仓位 (0~1) fee_rate: 单边费率, 万三即 0.0003 slippage: 单边滑点比例, 5bp 即 0.0005 lot: 最小交易单位, A 股为 100 股 """ cash, pos = init_cash, 0 equity = [] for i, dt in enumerate(bars.index): # 关键: t-1 日的信号在 t 日开盘执行, 天然规避未来函数 if i > 0 and bars["is_tradable"].iloc[i]: target = float(targets.iloc[i - 1]) px = bars["open"].iloc[i] buy_px, sell_px = px * (1 + slippage), px * (1 - slippage) nav = cash + pos * px target_shares = int(target * nav / buy_px / lot) * lot delta = target_shares - pos if delta > 0: cost = delta * buy_px * fee_rate cash -= delta * buy_px + cost pos += delta elif delta < 0: cost = -delta * sell_px * fee_rate cash -= delta * sell_px - cost # delta 为负, 卖出增加现金 pos += delta equity.append(cash + pos * bars["close"].iloc[i]) return pd.Series(equity, index=bars.index, name="equity")

几个参数需要说明。slippage单独给买卖两个方向，是因为真实成交里买入更容易成交在偏高价、卖出偏低价，用同一个中间价会系统性高估收益。lot取整会让小资金账户的目标仓位产生偏差——10 万本金买 20 元股票，一手 2000 元，5% 的仓位偏差就足以让回测和实盘对不上。fee_rate记得把印花税算进卖出边，A 股是单边收，很多人两边都按万三算，反而低估了成本。

至于为什么不在这里做「按收盘价成交」，是因为收盘价成交在实盘里几乎做不到——你要么在 14:57 之前下单，要么等到次日。用次日开盘价虽然也有偏差（跳空），但它是可复现的。

3. 三个能直接跑的定量策略：双均线、动量、均值回归

策略部分我建议按「先跑通一个最简单的，再谈优化」的顺序来。这份资源里给了不止一个示例策略，但真正值得你花时间的是三个原型：趋势跟随的双均线、截面动量、以及带门控的均值回归。它们的收益特征、换手率、失效环境完全不同，放在一起才能理解为什么「同一个参数在两只票上一个赚一个亏」。下面每个都给信号函数和参数取值区间，你可以直接替换进第 2 章的撮合循环。

3.1 双均线交叉：参数怎么取才不会自欺欺人

双均线的逻辑简单到不需要解释：快线上穿慢线满仓，下穿空仓。它的坑全在参数上。我见过太多人把 fast/slow 从 5 到 120 全网格搜一遍，挑出历史年化最高的那组，然后实盘三个月被打回原形。

比较稳妥的做法是先确定慢线周期，它对应你想捕捉的趋势级别：20 日大约是一个月，60 日是一个季度，120 日半年。选定之后，快线一般取慢线的 1/3 到 1/2，比如 20/60、10/30。这样参数不是搜出来的，是从「你打算持有多久」推出来的，可解释性完全不一样。

import pandas as pd def dual_ma_target(bars: pd.DataFrame, fast=20, slow=60) -> pd.Series: """返回目标仓位序列, 1 表示满仓, 0 表示空仓""" close = bars["close"] fast_ma = close.rolling(fast, min_periods=fast).mean() slow_ma = close.rolling(slow, min_periods=slow).mean() raw = (fast_ma > slow_ma).astype(float) # shift(1): 今天的均线状态明天才能用于下单 return raw.shift(1).fillna(0.0)

min_periods必须显式给，不然 rolling 默认在窗口不满时也返回 NaN，>比较后变 False，前 59 天全是空仓，回测起点就被悄悄挪后了。shift(1)是这段代码里唯一不能省的一行，它把「今天收盘算出的状态」推迟到明天生效，和第 2 章撮合层的targets.iloc[i-1]形成双重保险。真要只留一层的话，留在策略层更安全，因为别的撮合实现未必有这么严谨。

另外补一句：双均线在震荡市里的连续假信号是结构性缺陷，不是参数能修的。想缓解只能在信号之后加过滤，比如要求快慢线差值超过某个阈值才动手，或者用 ATR 做一道波动率门槛。这类过滤会降低换手，但也会削掉一部分趋势启动的头寸，取舍要看你账户能承受多少交易成本。

3.2 截面动量：调仓周期和手续费是一对冤家

截面动量的思路是定期给一篮子标的打分，买最强的若干个。它比双均线多两个参数：回看期和调仓周期。这两个参数对结果的敏感度远高于选股数量。

我一般把回看期设在 20 到 120 个交易日之间，跳过最近 5 天（避免短期反转效应污染动量）。调仓周期从 5 日、10 日、20 日里选，因为 A 股一周五个交易日，周频调仓的执行成本最容易估算。

import pandas as pd def cross_section_momentum(close_wide: pd.DataFrame, lookback=60, skip=5, top_n=3) -> pd.DataFrame: """ close_wide: 列是标的代码, 行是交易日的收盘价矩阵 返回每期各标的的目标权重 (等权持有 top_n 只) """ score = close_wide.shift(skip) / close_wide.shift(skip + lookback) - 1 rank = score.rank(axis=1, ascending=False) selected = rank <= top_n weights = selected.astype(float) weights = weights.div(weights.sum(axis=1).replace(0, 1), axis=0) return weights.fillna(0.0)

skip参数容易被忽略，但它是动量策略能否成立的关键之一：不跳过最近几天，你买进去的往往是刚拉完一波、正在回吐的标的。top_n不建议太小，3 到 5 只是兼顾分散和集中，低于 3 只时个股风险会盖过因子收益。返回的是权重矩阵而不是 0/1，是为了方便和撮合层配合——撮合层只关心「目标仓位是多少」，分批建仓的逻辑放在执行层处理。

还有一个隐性成本要算清楚：周频调仓意味着一年 50 次换手，双边万三加千一滑点，一年光成本就是 8% 左右。如果你的动量策略年化超额只有 10%，那基本是在给通道打工。这也是我建议先跑 20 日调仓、确认信号有效之后再缩短周期的原因。

3.3 均值回归：先做平稳性门控，再谈开仓

均值回归最反直觉的地方是：它不能无条件使用。价格序列本身不平稳，直接算 z-score 做多空，等于默认「偏离一定会回来」，而实际上一只股票可以从偏离 2 倍标准差一路走到 8 倍。

我的做法是在策略外面加一道门控：只有当标的的价差序列通过平稳性检验时才允许交易，不通过就空仓。这一步会让策略在很多时段完全不动作，但能避开大部分单边下跌里的接飞刀。

import numpy as np import pandas as pd def mean_revert_target(bars: pd.DataFrame, window=20, entry_z=2.0, exit_z=0.5) -> pd.Series: close = bars["close"] ma = close.rolling(window, min_periods=window).mean() sd = close.rolling(window, min_periods=window).std() z = (close - ma) / sd.replace(0, np.nan) pos = pd.Series(np.nan, index=close.index) pos[z <= -entry_z] = 1.0 # 低于下轨做多 pos[z >= entry_z] = 0.0 # 高于上轨清仓 pos[abs(z) <= exit_z] = np.nan # 回到中枢不动 # 用前值填充: 只有触发信号才改变仓位, 否则维持 return pos.ffill().fillna(0.0).shift(1)

这里entry_z和exit_z的差值就是你的持有区间，差值越小换手越高。window取 20 是因为它同时对应布林带的常见口径，和双均线的快线周期一致，方便你横向对比两套逻辑在同一段行情下的表现。注意sd可能为 0（连续横盘），replace(0, np.nan)是为了防止除零产生 inf，inf 会让后面所有比较失效。

三套逻辑的换手率差异很大，实盘前建议都过一遍同一个成本模型，再决定拿哪套做主力。

4. 从回测到全自动实盘：调度、状态与订单管理

回测跑通只是起点，真正让大多数人卡住的是「怎么让它每天自己跑」。这一层的技术难点不在策略，而在工程：定时触发、状态持久化、订单状态跟踪、异常兜底。任何一处偷懒，第二天早上你看到的可能就是重复下单或者仓位归零。

这份资源里实盘部分的结构值得借鉴：一个主循环负责按交易日历触发，一个执行器负责把目标仓位翻译成订单，一个状态文件负责在进程重启后恢复现场。三者之间靠文件或数据库通信，不共享内存。

4.1 主循环与状态持久化：断电也不能写坏状态

主循环的第一原则是幂等。同一个交易日跑十次，结果必须和跑一次一样。实现方式很简单：状态文件里记录「最后一次执行的交易日」，进入循环先比对，已经执行过就直接退出。

状态写入用「临时文件加原子替换」，而不是直接覆盖。直接覆盖的风险是写到一半进程被杀，留下一个半截的 JSON，下次启动解析失败，整个系统起不来。

import json, os STATE_FILE = "state.json" def load_state(path=STATE_FILE) -> dict: if not os.path.exists(path): return {"last_run_date": None, "position": 0, "open_orders": {}} with open(path, "r", encoding="utf-8") as f: return json.load(f) def save_state(state: dict, path=STATE_FILE) -> None: tmp = path + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(state, f, ensure_ascii=False) f.flush() os.fsync(f.fileno()) # 强制落盘, 防止系统缓存丢失 os.replace(tmp, path) # 原子替换, 要么旧要么新

os.replace在同一文件系统内是原子操作，这是这套写法成立的前提，所以临时文件必须和目标文件同目录。os.fsync在日频系统里其实可有可无，但如果你的程序跑在会随时被抢占重启的环境里，加上它更踏实。last_run_date用字符串存成YYYY-MM-DD，比较时直接字符串比对，省掉时区转换的麻烦。

4.2 订单状态机：别把「已提交」当成「已成交」

订单从提交到终态要经过一串中间状态，每个状态的处理方式不同。最常见的错误是把「已提交」当成「已成交」，然后立刻更新本地持仓，结果订单被拒，本地仓位和券商对不上，后面所有决策都是错的。

我一般维护一张订单表，字段至少包括订单号、标的、方向、委托数量、成交数量、状态、更新时间。状态只允许按固定路径流转，任何越级跳转都记warn日志。查不到终态时以券商回报为准，本地不猜。

ORDER_FLOW = { "pending": {"submitted", "rejected"}, "submitted": {"partial", "filled", "cancelled", "rejected"}, "partial": {"filled", "cancelled"}, "filled": set(), "cancelled": set(), "rejected": set(), } def transit(order: dict, new_status: str) -> dict: cur = order["status"] if new_status not in ORDER_FLOW.get(cur, set()): raise ValueError(f"非法状态流转: {cur} -> {new_status}") order["status"] = new_status return order

这个状态机最大的作用不是防错，而是让你在写日志时能一眼看出哪一步断了。实盘里我遇到过订单在submitted卡了二十分钟的情况，原因是对手方流动性太差、委托价挂在前一天的价位上，这种问题不看状态流转根本发现不了。filled和cancelled定义为终态且不可再转出，是为了避免重复回报把已成交订单改回挂单。

4.3 日志与监控：只记录你能用来复盘的字段

日志不是越多越好。我见过把每一行行情都打出来的实现，一天几个 G，真出问题时反而找不到关键那几行。有效的做法是分层：决策日志记录「为什么下单」，执行日志记录「下单后发生了什么」，两者用同一个run_id串起来。

决策日志至少要有：交易日、策略输出的目标仓位、当前实际仓位、差额、触发原因。执行日志至少要有：订单号、委托价、成交均价、滑点实测值。把滑点实测值和回测假设做对比，是判断策略是否需要重新标定的最快方式——如果连续两周实测滑点是假设的两倍，那回测收益基本可以打对折看。

监控指标建议就盯三个：当日净值、持仓数量、未完成订单数。任何一个和预期不符就发告警。告警渠道随便，邮件、webhook 都行，关键是别只写进日志文件——没人会天天翻日志，血泪经验。

5. 避坑清单：五个让回测漂亮、实盘翻车的地方

这一节是我认为整份资源里最该反复看的部分，也是我自己踩过最多次的地方。回测曲线越平滑，越要怀疑哪里出了问题。下面五条按「现象、原因、解决」写，每条都能独立对照检查。

5.1 未来函数：回测年化 40%，实盘两周回撤 15%

现象：回测收益高得离谱，胜率 70% 以上，净值几乎不回撤；实盘一上线就持续亏损，且亏损方向和回测完全相反。

原因：信号用到了当日收盘价甚至当日最高价，而撮合又假设按当日开盘价成交。等价于「用收盘后才知道的信息，在开盘时下单」。另一个隐蔽变种是用全样本的均值、标准差做标准化，比如(close - close.mean()) / close.std()，close.mean()已经包含了未来数据。

解决：所有滚动统计必须带min_periods并配合shift(1)；标准化一律改成滚动窗口内的rolling().mean()；最后做一次「截断测试」——只喂前 100 行数据跑一遍，再喂前 200 行跑一遍，如果前 100 行的信号在两次结果里不一致，说明有未来函数。

5.2 幸存者偏差：回测里全是活到今天的票

现象：股票池回测表现优异，但把同一套参数用在实时的全市场选股上，收益骤降。

原因：回测用的标的列表是「当前还在上市的公司」，退市的、被合并的、长期停牌的统统不在里面。等于在十年前就预知了谁会活下来。

解决：数据层必须保留历史成分股名单和退市记录，按「当时可交易的标的」构建股票池，退市日之后强制平仓。如果数据源拿不到历史成分，退而求其次的做法是把选股范围限制在指数成分内，并接受这不完美的近似。

5.3 成本假设：手续费填 0，滑点当不存在

现象：回测年化 25%，实盘年化 8%，差距稳定在十几个点，且换手越高差得越多。

原因：回测里手续费 0、滑点 0、按中间价全额成交。真实情况是每笔都要付佣金和印花税，大单还会冲击价格，小市值标的甚至买不到量。

解决：手续费按双边万三起算，卖出边加千一印花税；滑点按 5bp 起步，小市值标的给到 15bp；再加一条成交量约束——单日成交量不超过该标的当日成交额的 1%，超出部分拆单到次日。这三条加上去，很多策略的超额会消失，但消失的那些本来就不该做。

5.4 过拟合：参数网格搜出来的「最优点」

现象：网格搜索里最优参数年化 35%，相邻参数年化只有 5%，实盘用最优点表现接近随机。

原因：参数空间搜得够密，总能找到一组恰好拟合历史噪声的组合。这不是策略有效，是曲线拟合。

解决：看参数高原而不是参数尖峰。把 fast/slow 做成热力图，如果最优点周围一片绿，说明参数稳健；如果只有一个孤立的亮点，直接放弃。另外把样本切成训练段和验证段，训练段选参数、验证段只跑一次，验证段结果才是你要信的。

5.5 时钟与交易日历：程序在节假日也去下单

现象：节后第一天开盘出现重复委托，或者某个周一系统报「无行情数据」直接崩溃。

原因：主循环用固定sleep(86400)做日频调度，不认交易日历；服务器时区和交易所时区不一致；夏令时切换那天时间偏移一小时。

解决：调度一律以交易日历为准，非交易日直接跳过；所有时间戳统一存 UTC，只在展示层转本地时区；开盘前留一个「预检」步骤，检查行情是否就绪、账户是否可交易、状态文件里的last_run_date是否是上一个交易日，三项有任一不满足就不进入执行流程。

6. 用滚动窗口验证：把参数按时间切片，一次跑完

单次回测最大的问题是它只告诉你「这套参数在这段历史上表现如何」，没告诉你「换一段历史还成不成立」。我现在的习惯是任何策略上线前必须过一遍滚动验证（walk-forward）：把时间轴切成若干段，每段用前面的数据选参数、用后面的数据做测试，测试段的结果拼起来才是可信的净值曲线。

import itertools import pandas as pd def walk_forward(bars, signal_fn, param_grid, train=250, test=60, cost_fn=None): """ train: 训练窗口交易日数, 约一年 test: 测试窗口交易日数, 约一个季度 signal_fn: (bars, **params) -> 目标仓位序列 """ results = [] start = 0 while start + train + test <= len(bars): tr = bars.iloc[start:start + train] te = bars.iloc[start + train:start + train + test] best, best_score = None, -1e9 for combo in itertools.product(*param_grid.values()): params = dict(zip(param_grid.keys(), combo)) sig = signal_fn(tr, **params) score = evaluate(tr, sig) # 用夏普或卡玛比率均可 if score > best_score: best, best_score = params, score # 只用训练段选出的最优参数跑测试段, 不回头调参 sig_te = signal_fn(pd.concat([tr, te]), **best).iloc[train:] results.append({"period": te.index[0], "params": best, "test_equity": run_backtest(te, sig_te)}) start += test return results

三个参数决定这套验证的可信度。train取 250 个交易日，是因为大部分趋势策略需要至少一年的样本才能覆盖一轮完整波动；test取 60 个交易日，短于 40 天时单段结果噪声太大，长于 120 天又会让切分段数太少。param_grid里每一项不要超过 4 个候选值，乘积控制在 100 组以内，否则训练段本身就在过拟合。

跑完之后我一般看两件事：各测试段拼接起来的整体净值曲线，以及最优参数在各段之间的跳动幅度。曲线平稳、参数漂移小，说明策略捕捉的是某种持续存在的结构；曲线靠某一两段撑着、参数每段都换，那基本可以判定是拟合。

从那以后我每次要把策略从回测推到实盘，都会强制走一遍滚动验证加实测滑点对比，哪怕多花两天也不跳过——跳过的那两次，代价都是在实盘里用真金白银补的。希望帮到你。

本文还有配套的精品资源，点击获取