简介面向具备一定Python基础、希望系统掌握算法交易与量化策略实现的投资者和开发者这套资料围绕全自动交易系统的搭建流程展开从数据获取、技术指标计算、策略设计到回测、交易执行、风险控制与实时监控均有涉及适合作为量化交易入门与实战参考。压缩包共35个文件以34个Python脚本和1份Markdown说明为主整体约38KB脚本按功能拆分便于按主题查阅、复用与二次改造。已有3962人学习下载说明其在Python算法交易学习场景中具有一定关注度。读者可借助示例代码理解移动平均、RSI、布林带等指标计算以及回测框架、组合再平衡、绩效指标和常见交易策略的实现思路并参考日志记录、可视化监控与持续优化相关模块逐步搭建自己的自动化交易实验流程。1. 一套 Python 算法交易系统到底由哪几块拼起来有人拿着一份年化 60%、最大回撤 3% 的回测曲线来找我看代码问为什么按这条曲线搭起来的 Python 全自动交易系统实盘第一个月就吐回去一半。翻到信号函数第三行就明白了用当天收盘价算信号也用当天收盘价成交。这类问题在定量交易策略的落地过程中几乎是必经之路跟水平高低关系不大跟有没有把每一层职责拆干净关系很大。这份资料拆的是完整链路行情数据落成本地列式文件、信号与仓位分层、能表达撮合约束的回测引擎、以及把策略挂到定时任务上的调度骨架。它不是一份跑起来就能赚钱的模板而是一套让你能自己判断这条曲线可不可信的骨架。适合已经会用 pandas 做数据处理、想把 Excel 里的想法搬到代码里的人如果连 rolling 和 shift 都没用过建议先补两周基础再进来不然踩的坑会全是语法坑浪费时间。2. 行情数据落地把日线拉成能直接喂给回测的本地表数据层决定策略迭代速度。我见过最常见的低效做法是每次回测都去调一次接口拉全量历史——接口限频不说关键是历史数据会被修订昨天跑出来的夏普 1.8今天重跑变成 1.6你还找不到原因。把数据落成本地文件等于给自己留了一份可复现的黑匣子。2.1 为什么先把数据落成本地列式文件CSV 和 Parquet 的差别在单标的上看不出来标的池上到几百个、历史拉到十年以上差距就出来了。CSV 是行存你要读 close 一列它得把整行解析完再丢掉Parquet 是列存只读需要的列压缩比通常还能再省一半空间。回测里最频繁的操作是按日期切片、按列取价列存天然占优。我一般会把目录切成两层store/daily/放全市场日线按年份分区store/meta/放交易日历和标的基础信息。策略代码只认这两层不直接认任何接口接口换了只改一个 loader 文件。字段上建议固定成一套最小集合多一个字段都别加字段类型说明symbolstr(6)标的代码必须补零成字符串datedatetime64交易日不是自然日open/high/low/closefloat64不复权原始价volumefloat64成交量单位统一amountfloat64成交额用来算成交占比约束adj_factorfloat64复权因子单独存不提前乘进价格把复权因子单独存这一条值得展开说。前复权价格会随着新数据到来整体平移意味着你今天存的前复权价下个月就变了回测结果自然不可复现。后复权价格历史段不变可以直接用但如果你未来要接实盘对账真实成交价还是原始价。所以原始价和因子都留着用的时候现算是最省心的做法。2.2 复权、停牌与缺失值的处理脚本原始 CSV 通常来自不同来源列名五花八门第一步是显式映射绝对不能靠 rename 猜列。import pandas as pd from pathlib import Path RAW_DIR Path(./raw_csv) COL_MAP { 代码: symbol, 日期: date, 开盘价: open, 最高价: high, 最低价: low, 收盘价: close, 成交量: volume, 成交额: amount, } KEEP [symbol, date, open, high, low, close, volume, amount] def load_one(path: Path) - pd.DataFrame: # dtype 指定代码为字符串否则 000001 会被读成 1 df pd.read_csv(path, dtype{代码: str}, encodingutf-8) df df.rename(columnsCOL_MAP) df[symbol] df[symbol].str.zfill(6) # 显式给 format避免 pandas 把 20240101 猜成时间戳 df[date] pd.to_datetime(df[date], format%Y%m%d) # 数值列统一转 float64早期 CSV 里混进来的 - 会变成 object for c in [open, high, low, close, volume, amount]: df[c] pd.to_numeric(df[c], errorscoerce) return df[KEEP] frames [load_one(p) for p in sorted(RAW_DIR.glob(*.csv))] daily ( pd.concat(frames, ignore_indexTrue) .drop_duplicates(subset[symbol, date], keeplast) .sort_values([symbol, date]) .reset_index(dropTrue) ) daily.to_parquet(./store/daily/all.parquet, indexFalse)errorscoerce这一步不要省历史文件里经常混着--、空串、0.00这类占位符不转成 NaN 后面算收益率会直接抛异常。keeplast是给重复数据兜底通常是某天数据被补发过一次取后到的版本。复权计算我一般不在入库时做而是在取数函数里现算def to_backward_adjusted(df: pd.DataFrame) - pd.DataFrame: # 后复权历史价格不变最新价被放大回测可复现 out df.copy() for c in [open, high, low, close]: out[c] out[c] * out[adj_factor] return out停牌识别用价格和成交量两个条件一起判只看成交量为 0 会误伤集合竞价缺失的样本def mark_suspended(df: pd.DataFrame) - pd.DataFrame: out df.copy() # 最高价等于最低价且成交量为 0基本可以判定为全天无成交 out[suspended] (out[high] out[low]) (out[volume] 0) return out接下来是对齐交易日历这一步最容易出玄学问题。常见做法是拿一份交易日历对每个标的做reindex到全量日期ffill补价格、volume补 0同时保留suspended标记。不要用resample(B)工作日不等于交易日节假日缺的数据会被错误地补成前值导致连续多天的假信号。def align_to_calendar(df: pd.DataFrame, calendar: pd.DatetimeIndex) - pd.DataFrame: df df.set_index(date).reindex(calendar) df[suspended] df[suspended].fillna(True) # 日历里有但数据里没有 未上市或停牌 df[volume] df[volume].fillna(0.0) df[amount] df[amount].fillna(0.0) for c in [open, high, low, close, adj_factor]: # 价格用前值填避免出现 0 价导致的收益率爆炸 df[c] df[c].ffill() return df.reset_index().rename(columns{index: date})参数上我要提醒一句ffill之后必须保留suspended列回测撮合时拿它做过滤。只填价格不填标记等于让系统认为停牌期间还能正常成交这是后面第 5 章要专门讲的坑之一。数据层做到这里就够了再往下应该进策略层不要在数据层做任何信号预处理那会让信号和回测耦合在一起改一次参数要重跑全量清洗。3. 定量策略的代码化信号、仓位、风控拆成三层策略代码最常见的写法是一个大函数从头算到尾算均线、判断金叉、算仓位、扣手续费全在三十行里。这样写跑得通但改不动。换个标的池要重写加个波动率目标要重写做参数搜索更是没法批量跑。拆层不是为了好看是为了让每一层能被独立测试和替换。3.1 信号函数只负责想不想做信号层的输入只有历史价格窗口输出只有方向或打分不碰资金、不碰仓位、不碰成本。这一层越薄越好薄到你可以用三行单元测试验证它给定一段人造 K 线输出必须是预期的那几个值。import pandas as pd def signal_ma_cross(close: pd.Series, fast: int, slow: int) - pd.Series: 双均线方向信号1 表示看多0 表示空仓。只用历史数据。 ma_fast close.rolling(fast, min_periodsfast).mean() ma_slow close.rolling(slow, min_periodsslow).mean() sig (ma_fast ma_slow).astype(float) # min_periods 保证前 slow-1 根是 NaN而不是被当成 0 信号 sig[ma_slow.isna()] float(nan) return sigmin_periods这个小参数是血泪经验。默认rolling(60)在前 59 根会返回 NaN但如果你的代码里顺手加了个.fillna(0)前 59 根就变成明确看空回测开局会出现一段并不存在的空仓期年化算出来偏高。信号层输出的 NaN 应该由仓位层决定怎么处理而不是在这里抹平。信号层还有个约定不在这里做 shift。shift(1)是回测层的事放在信号层会让这个信号到底用了几号的数据变得模糊而模糊正是未来函数最喜欢的藏身处。3.2 仓位与风控函数负责做多少仓位层接收信号和波动率输出目标权重。风控也放在这一层因为风控本质上是给权重加约束单票上限、总仓位上限、回撤熔断。import numpy as np def position_by_vol( ret: pd.Series, sig: pd.Series, vol_window: int 20, target_vol: float 0.15, max_weight: float 0.1, ) - pd.Series: # 年化已实现波动252 是交易日数 rv ret.rolling(vol_window, min_periodsvol_window).std() * np.sqrt(252) # 波动越大仓位越小上限截断防止低波期加杠杆 scale (target_vol / rv.replace(0, np.nan)).clip(upper1.0) w (sig * scale).clip(lower0.0, uppermax_weight) return w.fillna(0.0)target_vol / rv这里有一个隐含假设波动率是可预测的。实际并不完全成立所以clip(upper1.0)必须有否则低波区间会把仓位放大到两倍三倍回测里看着很赚实盘遇到波动跳升直接爆仓。想更稳一点可以在rv上再做一次ewm平滑代价是信号滞后一两天。风控层再往上还有一个组合级约束我一般单独写成一个函数在权重矩阵上按行处理def apply_portfolio_limits(w: pd.DataFrame, max_total: float 1.0, max_per_symbol: float 0.1) - pd.DataFrame: w w.clip(uppermax_per_symbol) total w.sum(axis1) # 总仓位超限时按比例缩放而不是直接砍掉某几只 scale (max_total / total).clip(upper1.0) return w.mul(scale, axis0)按比例缩放比直接截断好截断会让权重靠后的标的永远拿不到仓位组合集中度莫名其妙地偏高。3.3 参数化与网格搜索的写法参数用 dataclass 装起来好处是能直接 dump 成 json 存进结果表回看时知道这组曲线对应哪组参数不用翻 commit 记录。from dataclasses import dataclass, asdict from itertools import product dataclass class Params: fast: int 20 slow: int 60 vol_window: int 20 target_vol: float 0.15 max_weight: float 0.1 grid {fast: [5, 10, 20], slow: [40, 60, 120], vol_window: [10, 20, 40]} combos [dict(zip(grid.keys(), v)) for v in product(*grid.values())] combos [c for c in combos if c[fast] c[slow]] # 过滤无效组合fast slow这条过滤看着 trivial但能砍掉三分之一的无意义回测。跑完之后别急着看夏普排名先看参数平原把夏普按 fast 和 slow 画成热力图如果最优值是一座孤峰周围一圈立刻掉到 0.3这组参数基本是过拟合如果是一片缓坡说明策略对这个参数不敏感换到样本外还有机会活。样本切分我一般用滚动前推前 3 年训练、后 1 年验证、窗口向前滚。单次切分比如 8 年训练 2 年验证只能给出一个点估计滚动前推能给出十来个点看的是策略在不同市场状态下的稳定性而不是某一年的运气。4. 回测引擎落地手续费、滑点、撮合时点怎么建模回测引擎的职责只有一个尽量诚实地模拟如果当时真的这么做会发生什么。它不需要多复杂但必须能表达三件事——你什么时候下单、以什么价格成交、成交不了怎么办。表达不了这三件事的引擎跑出来的数字再好看也只是曲线拟合。4.1 向量化回测与事件驱动回测的边界向量化回测就是整条收益率序列一次性算完(w.shift(1) * ret).sum(axis1)这种写法。优点是快几百组参数几分钟跑完缺点是无法表达资金约束和撮合约束。它表达不了的东西具体包括可用资金不足时只能部分成交、单票单日成交量占比超过上限买不进、涨跌停板上的单子成交概率极低、分红送股导致的持仓数量变化。这些约束在低频宽基策略里影响可能只有几个点但在高换手中小市值策略里能吃掉全部超额收益。我一般走两段式参数初筛用向量化从几百组里挑出夏普前 20 到 30 组然后用事件驱动引擎重跑这 20 组看排名变化。如果排名大幅洗牌说明策略对成本或撮合极度敏感这时候直接放弃比调参更明智。这个流程能在可接受的时间里把看着能赚和可能能赚区分开。4.2 一个可复现的最小事件驱动回测循环下面这个循环刻意写得笨逐日遍历不追求速度。它的价值在于每一行你都能看懂出问题能定位。import numpy as np import pandas as pd def run_backtest( open_px: pd.DataFrame, # index交易日, columns标的, T日开盘价 target_w: pd.DataFrame, # index交易日, columns标的, T日收盘后算出的目标权重 commission: float 3e-4, # 单边佣金率 slippage: float 1e-3, # 单边滑点率 min_fee_rate: float 5e-4 # 最低佣金折算成的最小费率防止小额交易被低估 ) - pd.DataFrame: dates open_px.index nav, rows 1.0, [] weights pd.Series(0.0, indexopen_px.columns) for i in range(1, len(dates) - 1): cur_d, nxt_d dates[i], dates[i 1] # 昨天收盘算出的目标今天开盘执行 want target_w.loc[dates[i - 1]].fillna(0.0) # 1) 调仓成本换手率 × (佣金 滑点) turnover (want - weights).abs().sum() cost turnover * max(commission slippage, min_fee_rate) nav * (1.0 - cost) # 2) 持有到下一个开盘收益用开盘价对开盘价 ret (open_px.loc[nxt_d] / open_px.loc[cur_d] - 1.0).fillna(0.0) nav * (1.0 float((want * ret).sum())) rows.append({date: cur_d, nav: nav, turnover: float(turnover)}) weights want return pd.DataFrame(rows).set_index(date)关键在dates[i - 1]和dates[i]这个错位。目标权重来自昨天收盘后的数据成交发生在今天开盘这两个日期差一天是整段代码里唯一防未来函数的地方。所有回测收益离谱地好的问题八成都是这个错位被写成了对齐。min_fee_rate这个参数容易被忽略。真实佣金有最低收费比如一笔几百块的交易也收固定 5 元折算成费率可能高达 1%。如果你的策略持仓分散、单笔金额小按纯比例算佣金会严重低估成本。用max(commission slippage, min_fee_rate)做一个粗略上界比精确建模简单得多效果也够。再往下就是成交约束了。真实回测里我会在调仓前加一道过滤# volume 为当日成交量max_participation 通常取 0.05 ~ 0.1 cap volume.loc[cur_d] * max_participationmax_participation取多大取决于你的资金量和标的流动性。资金量小的时候这个约束几乎不触发规模上去之后它会成为主要瓶颈。我一般会用两三个不同的取值跑一遍看净值曲线的衰减速度这比单纯调策略参数更能告诉你这套东西的资金容量上限。回测输出除了净值一定要把turnover序列留下来。换手率是判断策略性质的最快指标年换手 50 次和 5000 次的策略对成本、滑点、撮合的要求完全是两个量级用同一套参数评估没有意义。5. 避坑与排查回测漂亮实盘翻车的五个高频原因5.1 未来函数藏在 shift 的那一行现象回测夏普 3 以上净值曲线平滑得不像真的实盘上线后回撤立刻扩大三倍。原因信号和成交用了同一天的收盘价。(signal * ret).sum()这种写法里signal是 T 日收盘算出来的ret也是 T 日收盘对 T-1 日收盘的收益等于用当天收盘的信息去赚当天的钱。这类错误在向量化回测里几乎是默认写法不写shift(1)就不会发现。解决统一约定T 日收盘生成信号、T1 开盘成交向量化里写成(signal.shift(1) * ret_open_to_open).sum()事件驱动里写成target_w.loc[dates[i-1]]配合open_px.loc[dates[i]]。改完再跑一遍如果夏普从 3 掉到 0.8不要沮丧0.8 才是这条策略的真实起点。5.2 复权方式选错回测结果不可复现现象同一份代码、同一份数据隔一个月重跑年化收益从 18% 变成 24%找不到任何代码改动。原因用了前复权价格。前复权以最新价为基准往回推只要这段时间里标的有过分红送股历史价格就整体平移历史信号跟着变。你上次跑的结果是建立在当时的最新价这个已经消失的基准上的。解决存原始价 复权因子回测时现算后复权价格或者退一步用后复权价。判断方法很简单把数据区间截到现在往前推半年同一段历史价格两次读取必须完全一致不一致就是基准被动过。从那以后我每次建库都会跑一个历史价格一致性校验比事后查数据省事得多。5.3 参数网格挑出来的最优是过拟合现象参数搜索里某个组合夏普 2.6旁边一圈全是 0.4 到 0.7换成样本外直接负收益。原因在样本内做了穷举搜索然后挑了最高点。搜索次数越多最高点里包含的噪声越大这在参数平原上表现为一座孤峰。孤峰不是策略有效是噪声恰好堆在这里。解决看参数平原而不看单点要求最优点的邻域内至少有一半参数组合夏普大于 0.8做滚动前推看十几次切分里胜率有没有过六成把参数搜索的总次数记下来次数上到三位数就要提高警惕。我一般会把网格控制在 50 组以内宁可少试几组也不要在噪声上建房子。5.4 成本建模偷懒换手率一高就露馅现象低频版本和实盘对得挺准把调仓频率从月度改成周度回测年化从 15% 冲到 40%实盘只有 6%。原因只算了佣金没算滑点和冲击成本。换手率翻五倍滑点成本也跟着翻五倍而且真实市场里这个关系不是线性的资金量越大越差。回测里按固定 bps 算滑点等于假设你的单子永远能按中间价成交。解决滑点按换手率分档加压测比如 1bp、5bp、15bp 各跑一遍看净值曲线的形状还在不在。同时加成交量占比约束单票单日不超过成交量的 5%。如果一个策略只有把滑点压到 1bp 以下才赚钱那它在真实市场里就是不赚钱。5.5 涨跌停和停牌让回测买进了买不到的票现象回测里某天按开盘价买进了一只票实盘挂单一天没成交第二天高开一大截。原因撮合环节没有排除不可成交状态。停牌期间价格被前值填充看起来有价格实际一分钱都换不了手涨跌停板上的开盘价是真实价格但那个价格上根本没有卖单给你。解决在撮合前加两道过滤suspended为 True 的标的当日目标权重强制置 0用开盘价相对昨收的涨跌幅判断是否触及板位触及就把成交概率降到 0 或只允许部分成交。这个过滤会让回测收益下降但降下来的那部分本来就是假的。做完这一步回测和实盘的差距通常会从完全对不上缩小到几个点的合理误差。6. 全自动调度把策略接到定时任务与告警上回测跑通只是到了半山腰真正让系统变成全自动的是调度。这一段代码量不大但出问题的影响最直接任务没跑、跑了两次、跑完写库失败任何一种都会让持仓和预期对不上。6.1 幂等写入与任务清单所有落库操作按(trade_date, strategy_id, symbol)做主键 upsert不要用 insert。调度框架重试是常态非幂等写入会让同一天的信号出现两条记录下游按最新一条读看起来正常实际已经错了。一个最小任务清单大概是这个形态任务触发时间输入输出失败处理数据增量更新收盘后 18:00数据接口daily.parquet重试 3 次后告警信号计算数据就绪后daily.parquet 参数signal 表中止当日流程组合优化信号就绪后signal 表target_weight 表回退到上一版权重下单执行次日开盘前target_weight 表委托记录逐笔核对不一致告警对账收盘后委托记录 成交回报持仓差异表差异超阈值告警依赖关系用数据就绪标记串不要用固定时间间隔串。数据接口晚到十分钟后面所有任务都该顺延硬编码 sleep 迟早出事。6.2 失败告警与自愈告警要能回答三个问题哪个任务、什么时间、失败原因是什么。日志里只写task failed等于没报警。我一般会在每个任务的入口和出口各打一条结构化日志包含任务名、交易日、耗时、记录条数失败时把异常栈单独落一份文件。import logging, traceback from datetime import datetime def run_task(name: str, trade_date: str, fn): t0 datetime.now() logging.info(f[START] {name} {trade_date}) try: n fn(trade_date) logging.info(f[OK] {name} {trade_date} rows{n} cost{(datetime.now()-t0).seconds}s) except Exception: logging.error(f[FAIL] {name} {trade_date}\n{traceback.format_exc()}) # 关键任务失败直接抛让调度器走重试或熔断 raise自愈只做一件小事数据更新失败重试三次第三次仍失败就发告警并跳过当日交易绝不带着不完整数据下单。这条规则我吃过亏——某次接口返回了半天的数据程序以为是完整交易日算出来的信号全是下午的第二天开盘按错误信号调了一遍仓。从那以后我每次接入新的数据源都强制走一遍记录条数 最大日期 缺失标的数三道校验任何一道不过就熔断。调度这一层的价值不在代码在于它把策略和运维分开了。策略该关心的是信号质量运维该关心的是任务有没有跑成功、数据有没有到齐、订单有没有对上。混在一起写最后两边都做不好。希望帮到你。本文还有配套的精品资源点击获取