简介一套面向金融量化初学者及具备Python基础开发者的算法交易实战代码集围绕全自动交易系统的搭建串联数据获取、技术指标计算、策略回测、交易执行与风险控制等环节帮助读者把零散的Python金融知识组织成可运行的策略流程。包内共35个文件以34个.py脚本和1个README.md说明文档为主压缩包约38KB脚本按功能拆分涵盖均线、RSI、MACD、ADX、布林带与ATR等指标实现夏普与索提诺比率、最大回撤与卡尔玛比率等绩效度量以及组合再平衡、突破策略、神奇公式选股、财务数据抓取与可视化监控等模块README提供整体索引。已有3962人学习下载。读者可据此理解量化策略从数据清洗、指标计算到回测评估的完整实现路径并借鉴其中的函数组织方式与多数据源接口脚本快速搭建自己的研究框架。1. 从一次凌晨的止损说起这套 Python 算法交易系统值不值得拆凌晨两点行情突然拉出一根长阳线你从床上爬起来手动平仓手指还没点下去价格又回去了——这是不少做算法交易的人共有的一段记忆。这个用 Python 搭起来的全自动交易系统解决的正是这件事把行情采集、策略信号、仓位计算、下单执行、事后对账串成一条无人值守的流水线让定量交易策略在回测和实盘里跑同一套代码逻辑。它适合会一点 Python、想把交易想法落成工程代码的开发者也适合手里已经有策略、但还没搭过完整骨架的人。整套代码以 Python 为主用 pandas 处理行情事件驱动做回测调度器定时触发配置文件切换模拟盘与实盘。2. 系统骨架行情到订单的六层结构与目录落地先把「能跑」这件事拆开看。一套全自动交易系统最容易翻车的地方不是策略写得多花哨而是层次没分清楚策略直接调下单接口风控塞在策略里回测和实盘各写一份逻辑最后两边行为对不上。这个项目的做法是把职责切成六层每层只认上下相邻层的输入输出回测和实盘共用中间四层只在最外侧换执行通道。2.1 六层分层与一次完整的数据流六层分别是配置层、数据层、策略层、风控层、执行层、监控层。配置层贯穿全局负责把标的、频率、参数、资金上限从 YAML 文件读进内存数据层负责拉行情、清洗、落本地库策略层只做一件事吃进一根 bar吐出信号或 None风控层拿到信号后决定「做不做、做多少」执行层负责把目标仓位翻译成订单、发出去、收回报监控层写日志、算权益、发告警。一次完整的链路是这样跑的调度器在预定时间唤醒任务数据层先补齐本地缺失的 bar策略层用最近 N 根 bar 计算指标产出Signal(symbol, side, strength, reason)风控层结合当前持仓和当日盈亏给出目标数量执行层对比目标与现有持仓的差额生成订单并落库成交回报回来后更新持仓和权益监控层把这一轮的关键字段写进日志。之所以要拆出「目标仓位」这一层而不是让策略直接说「买 100 股」是因为回测和实盘的资金曲线不一样同一个信号在不同权益下对应的手数不同。策略只管方向数量交给风控统一算这样换标的、换资金规模时不用改策略代码。提示策略层返回None表示「不动作」返回信号表示「希望调整到某个方向」。不要让它返回具体股数否则后面接第二个执行通道时必然要重写。2.2 目录结构与依赖清单目录按层切分一个包一层测试目录单独放。这样做的直接好处是排查问题时能快速定位行情不对去data/信号不对去strategy/下单不对去execution/不会在几千行代码里翻。quant-trader/ ├── app/ │ ├── main.py # 命令行入口按 mode 分发 │ ├── scheduler.py # 定时任务模拟盘/实盘共用 │ └── context.py # 组装各层对象类似依赖注入容器 ├── config/ │ └── config.yaml # 标的、频率、参数、风控阈值 ├── data/ │ ├── feed.py # 行情拉取与增量更新 │ ├── store.py # SQLite 读写 │ └── normalize.py # 列名、时区、类型统一 ├── strategy/ │ ├── base.py # Strategy 基类与 Signal 定义 │ ├── dual_ma.py # 双均线示例策略 │ └── registry.py # 策略注册表按名字取类 ├── backtest/ │ ├── engine.py # 事件驱动回测循环 │ ├── broker.py # 模拟撮合 │ └── metrics.py # 绩效指标 ├── risk/ │ └── manager.py # 仓位计算与熔断 ├── execution/ │ ├── broker.py # 执行通道抽象 │ ├── paper.py # 模拟盘实现 │ └── live.py # 实盘实现对接具体网关 ├── monitor/ │ ├── logger.py # 结构化日志 │ └── alert.py # 异常告警 └── tests/ ├── test_engine.py └── test_risk.py依赖不多核心就几个。版本不用追新能锁就锁回测结果对不上时第一步就是核对依赖版本。依赖用途备注pandas行情清洗、指标计算主力工具别在循环里逐行 applynumpy数值运算指标向量化时用SQLite3本地行情与订单落库标准库自带单机足够APScheduler定时调度模拟盘和实盘共用一套PyYAML配置文件解析参数外置别写死在代码里pytest单元测试回测引擎和风控必须有测试行情拉取那一层建议做成可替换的适配器定义一个MarketGateway抽象类具体实现放子类里接口只保留get_bars(symbol, freq, start, end)一个方法。换数据源时只改一个文件策略和回测完全不用动。2.3 配置与运行入口一个命令行切三种模式入口只做三件事解析参数、加载配置、按模式分发。回测、模拟盘、实盘走同一个build_context区别只在最后注入的执行通道。# app/main.py import argparse from app.context import build_context from execution.paper import PaperBroker from execution.live import LiveBroker from backtest.engine import BacktestEngine def parse_args(): p argparse.ArgumentParser() p.add_argument(--mode, choices[backtest, paper, live], requiredTrue) p.add_argument(--config, defaultconfig/config.yaml) p.add_argument(--strategy, defaultdual_ma) return p.parse_args() def main(): args parse_args() ctx build_context(args.config, args.strategy) # 数据层、策略、风控都在这里装配好 if args.mode backtest: engine BacktestEngine(ctx.data, ctx.strategy, ctx.risk, ctx.backtest_broker) result engine.run() print(result.summary()) # 年化、回撤、夏普、成交笔数 elif args.mode paper: from app.scheduler import run_live run_live(ctx, brokerPaperBroker(ctx.cfg)) else: from app.scheduler import run_live run_live(ctx, brokerLiveBroker(ctx.cfg)) if __name__ __main__: main()--mode决定用哪个撮合器--strategy从注册表里取类--config指定参数文件。这样做的意义在于回测和实盘跑的是同一个strategy.on_bar同一套风控同一个订单生成逻辑唯一变量是撮合方式。很多「回测赚钱实盘亏钱」的问题根子就在这两条路径各写了一份代码。build_context里做的事不复杂读 YAML、初始化数据库、构造策略实例、构造风控实例。把它单独抽出来是为了测试时能直接传一个假配置进去不用启动整个程序。# config/config.yaml symbols: [DEMO] freq: 1d start: 2022-01-01 end: 2024-12-31 cash: 100000 strategy: dual_ma: fast: 5 slow: 20 risk: max_position_pct: 0.2 # 单标的最大占用权益比例 max_total_exposure: 0.8 # 总仓位上限 stop_loss_pct: 0.03 # 单笔止损 max_daily_loss_pct: 0.02 # 当日亏损熔断线参数全部外置的好处是调参时不用改代码也就不容易顺手改错逻辑。cash只影响回测初始资金实盘里以账户实际权益为准代码里用broker.equity()统一取值。3. 数据层与回测引擎把「未来函数」挡在门外回测结果漂不漂亮八成取决于数据层和撮合口径。这一章拆三个地方行情怎么增量落库、回测循环怎么保证策略看不到未来、绩效指标怎么算才不会被自己骗。3.1 行情采集与本地缓存行情不能每次启动都全量拉既慢又容易触发接口限流。做法是本地建一张 bar 表以(symbol, freq, ts)做联合主键每次只补最新一根之后的数据。INSERT OR REPLACE保证同一条 bar 重复拉取时不会产生脏数据。# data/store.py import sqlite3 import pandas as pd from contextlib import contextmanager DB_PATH data/market.db contextmanager def conn(): c sqlite3.connect(DB_PATH) try: yield c c.commit() finally: c.close() def init_db(): with conn() as c: c.execute( CREATE TABLE IF NOT EXISTS bars ( symbol TEXT NOT NULL, freq TEXT NOT NULL, ts TEXT NOT NULL, open REAL, high REAL, low REAL, close REAL, volume REAL, PRIMARY KEY (symbol, freq, ts) )) c.execute( CREATE TABLE IF NOT EXISTS orders ( client_id TEXT PRIMARY KEY, -- 幂等键重复提交直接被拒 symbol TEXT, side TEXT, qty REAL, status TEXT, created_at TEXT )) def upsert_bars(df: pd.DataFrame, symbol: str, freq: str) - int: rows [(symbol, freq, r.ts, r.open, r.high, r.low, r.close, r.volume) for r in df.itertuples()] with conn() as c: c.executemany(INSERT OR REPLACE INTO bars VALUES (?,?,?,?,?,?,?,?), rows) return len(rows) def last_ts(symbol: str, freq: str): with conn() as c: row c.execute(SELECT MAX(ts) FROM bars WHERE symbol? AND freq?, (symbol, freq)).fetchone() return row[0]conn()用上下文管理器包住异常时自动回滚避免半个事务留在库里。orders表的主键是client_id这个设计后面讲幂等下单还会用到同一个客户端订单号只能插一次网络重试时第二次插入直接失败从库层面挡住重复下单。增量拉取的逻辑放在feed.py核心是先查本地最新时间戳再决定请求区间# data/feed.py import pandas as pd from data.store import last_ts, upsert_bars FREQ_MINUTES {1m: 1, 5m: 5, 15m: 15, 1h: 60, 1d: 1440} def fetch_incremental(gateway, symbol: str, freq: str, start: str, end: str) - int: last last_ts(symbol, freq) if last: step pd.Timedelta(minutesFREQ_MINUTES[freq]) start (pd.Timestamp(last) step).isoformat() if pd.Timestamp(start) pd.Timestamp(end): return 0 raw gateway.get_bars(symbol, freq, start, end) # 统一网关屏蔽通道差异 df normalize(raw) # 列名、时区、类型对齐 return upsert_bars(df, symbol, freq)normalize这一步不能省不同数据源返回的列名可能是datetime也可能是time时间戳可能是带时区的也可能是本地时间成交量单位可能是股也可能是手。统一成ts/open/high/low/close/volume六列、UTC 时间戳、float 类型后面所有计算才有可比性。注意日线数据的ts建议统一用「bar 结束时间」。如果用开始时间回测里很容易出现「当天收盘价在当天开盘前就能看到」的错觉这类错误肉眼几乎看不出来。3.2 事件驱动回测循环回测引擎的关键约束只有一条策略在处理第 i 根 bar 时只能访问第 0 到第 i 根的数据。写法上有两种一种是把整个 DataFrame 传给策略让它自己切片一种是每根 bar 单独喂进去。前者灵活但容易写出未来函数后者啰嗦一点但安全。这个项目选后者。# backtest/engine.py import pandas as pd class BacktestEngine: def __init__(self, data: pd.DataFrame, strategy, risk, broker, cash100_000): self.data data self.strategy strategy self.risk risk self.broker broker self.broker.init_cash(cash) self.strategy.reset() def run(self): for i, (ts, bar) in enumerate(self.data.iterrows()): self.broker.on_bar(ts, bar) # 先撮合上一根 bar 挂出的单 if i self.strategy.warmup(): # 指标预热期不发信号 continue signal self.strategy.on_bar(ts, bar) if signal is None: continue price bar[close] target self.risk.size( signalsignal, equityself.broker.equity(price), priceprice, current_qtyself.broker.position(signal.symbol), ) self.broker.submit(ts, signal.symbol, target, bar, reasonsignal.reason) return self.broker.result()循环里的顺序很重要先让撮合器处理上一根 bar 产生的挂单再让策略基于当前 bar 出信号最后把新订单挂出去。这样成交价永远是「下一根 bar 的价格」和实盘里「信号产生后下一时刻才能成交」的时序一致。如果反过来先出信号再撮合等于用当前 bar 的收盘价成交回测收益会被系统性高估。warmup()是策略自己声明的预热长度双均线策略里等于慢线周期加一。预热期跳过信号避免用不足窗口的均线做判断。模拟撮合器backtest/broker.py需要处理三件事按开盘价成交、扣手续费和滑点、更新持仓和权益。滑点别设成 0哪怕只加万一也能让回测结果更接近真实。手续费按成交额比例扣卖出时额外扣印花税之类的费用具体比例放配置里。3.3 绩效指标别只看总收益回测跑完输出的不能只有一条资金曲线。最少要有年化收益、最大回撤、夏普比率、胜率、盈亏比、换手次数。指标计算本身不难难的是口径统一年化按 252 个交易日还是按实际天数夏普用日收益还是周收益最大回撤用权益还是净值这些要在代码里写死不能每次手算。# backtest/metrics.py import numpy as np import pandas as pd def max_drawdown(equity: pd.Series) - float: peak equity.cummax() dd equity / peak - 1.0 return float(dd.min()) # 返回负数-0.15 表示回撤 15% def sharpe(returns: pd.Series, periods: int 252) - float: std returns.std() if std 0 or np.isnan(std): return 0.0 return float(returns.mean() / std * np.sqrt(periods)) def summarize(equity: pd.Series, trades: pd.DataFrame) - dict: returns equity.pct_change().dropna() days (equity.index[-1] - equity.index[0]).days or 1 total equity.iloc[-1] / equity.iloc[0] - 1 annual (1 total) ** (365 / days) - 1 wins trades[trades[pnl] 0] return { total_return: round(total, 4), annual_return: round(annual, 4), max_drawdown: round(max_drawdown(equity), 4), sharpe: round(sharpe(returns), 2), trade_count: len(trades), win_rate: round(len(wins) / len(trades), 3) if len(trades) else 0.0, profit_factor: round(wins[pnl].sum() / abs(trades[trades[pnl] 0][pnl].sum()), 2) if (trades[pnl] 0).any() else float(inf), }max_drawdown用累计最大值做基准返回负值阅读时直接理解成「最大回撤 15%」。sharpe里periods252对应日频如果回测用的是小时线要改成对应周期数否则夏普会虚高。profit_factor是总盈利除以总亏损的绝对值小于 1 说明这套策略在样本内就没赚钱。指标出来之后至少要做一次分段检验把回测区间切成前后两半看两段的年化和回撤是否接近。如果前半段年化 40%、后半段年化 3%大概率是参数拟合了某一段行情后面第 5 章会细说。4. 策略与风控信号、仓位、止损三件套怎么接策略层和风控层是整个系统里改动最频繁的两块。策略负责「看不看得上」风控负责「吃不吃得下」。这两层之间用Signal这个数据类传递字段少但每个都有用。4.1 策略基类与双均线实现基类只定义三个方法warmup()返回预热长度reset()清空内部状态on_bar(ts, bar)返回信号或 None。信号里带reason字段是为了事后复盘时能直接看出这笔单子为什么下不用再翻代码重算指标。# strategy/base.py from dataclasses import dataclass from typing import Optional dataclass class Signal: symbol: str side: str # buy / sell strength: float # 0~1用于缩放目标仓位 reason: str # 写进日志和订单备注 class Strategy: def __init__(self, params: dict): self.params params def warmup(self) - int: return 0 def reset(self): pass def on_bar(self, ts, bar) - Optional[Signal]: raise NotImplementedError双均线是最小可运行示例逻辑是快线上穿慢线买入、下穿卖出。真实策略当然不会这么简单但用它验证整条链路是否通畅足够了。# strategy/dual_ma.py from strategy.base import Strategy, Signal class DualMA(Strategy): def reset(self): self.history [] self.holding False def warmup(self) - int: return self.params[slow] 1 def on_bar(self, ts, bar): self.history.append(float(bar[close])) if len(self.history) self.warmup(): return None fast_n, slow_n self.params[fast], self.params[slow] fast sum(self.history[-fast_n:]) / fast_n slow sum(self.history[-slow_n:]) / slow_n if fast slow and not self.holding: self.holding True return Signal(bar[symbol], buy, 1.0, fMA{fast_n}{slow_n}) if fast slow and self.holding: self.holding False return Signal(bar[symbol], sell, 1.0, fMA{fast_n}{slow_n}) return Nonehistory只追加不裁剪在长回测里会吃内存实际项目里可以改成固定长度的deque(maxlenslow_n 5)。holding是策略内部对持仓的认知只用来避免重复发同方向信号真正的持仓以执行层回报为准两者不一致时以执行层为准这个差异在实盘里会真实出现。策略注册表用一个字典按名字映射到类新增策略只需加一行注册不用改入口# strategy/registry.py from strategy.dual_ma import DualMA REGISTRY { dual_ma: DualMA, } def create(name: str, params: dict): if name not in REGISTRY: raise KeyError(f未知策略: {name}) return REGISTRY[name](params)4.2 仓位管理与风控拦截风控层的输入是信号、当前权益、当前价格、现有持仓输出是目标持仓数量。中间要过四道闸单标的仓位上限、总仓位上限、单笔止损、当日亏损熔断。# risk/manager.py class RiskManager: def __init__(self, cfg: dict, broker): self.max_pos_pct cfg[max_position_pct] self.max_total_pct cfg[max_total_exposure] self.stop_loss_pct cfg[stop_loss_pct] self.max_daily_loss_pct cfg[max_daily_loss_pct] self.broker broker def size(self, signal, equity, price, current_qty) - int: # 闸门一当日亏损超过阈值拒绝一切开仓 if self.broker.daily_pnl_pct() -self.max_daily_loss_pct: return current_qty if signal.side sell: return 0 # 清仓交给执行层按差额下单 # 闸门二单标的仓位上限 max_value equity * self.max_pos_pct * signal.strength target_qty int(max_value / price) # 闸门三总仓位上限超出部分等比压缩 used self.broker.total_exposure(price) room equity * self.max_total_pct - used if room 0: return current_qty target_qty min(target_qty, int(room / price)) # 闸门四已有持仓且浮亏触线先平再谈 if current_qty 0 and self.broker.unrealized_pct(signal.symbol, price) -self.stop_loss_pct: return 0 return max(target_qty, 0)四个参数的作用要分清max_position_pct管单只标的占总权益的比例防止一次押太重max_total_exposure管所有持仓加起来的上限防止满仓stop_loss_pct是单笔止损浮亏超过就清仓max_daily_loss_pct是当日熔断亏到线就当天下班。建议起步值分别是 0.2、0.8、0.03、0.02跑顺了再按策略波动率调整。size返回的是「目标数量」而不是「本次买卖数量」差额由执行层算。这个约定很关键信号重复触发时目标数量不变执行层算出来的差额是 0自然不会重复下单。注意熔断之后不要用「明天自动恢复」这种隐式逻辑最好在配置里显式写一个恢复时间或者干脆要求人工确认后重启任务。自动恢复碰上连续亏损行情会把小回撤拖成大回撤。4.3 订单执行与幂等网络抖动时的后悔药执行层做三件事把目标数量翻译成买卖差额、生成幂等订单号、处理成交回报和超时。最难的是第三件因为网络超时的时候你根本不知道单子到底成没成。# execution/paper.py import hashlib from data.store import conn class PaperBroker: def __init__(self, cfg): self.cfg cfg self.positions {} self.cash cfg.get(cash, 100_000) def submit(self, ts, symbol, target_qty, bar, reason): current self.positions.get(symbol, 0) diff target_qty - current if diff 0: return None side buy if diff 0 else sell client_id hashlib.md5( f{symbol}|{ts}|{side}|{abs(diff)}.encode() ).hexdigest()[:16] with conn() as c: try: c.execute( INSERT INTO orders VALUES (?,?,?,?,?,?), (client_id, symbol, side, abs(diff), pending, str(ts)), ) except Exception: return client_id # 主键冲突说明这笔单已经提交过 fill_price bar[open] if bar[open] else bar[close] # 下一根开盘成交 self._apply_fill(client_id, symbol, side, abs(diff), fill_price, ts) return client_id def _apply_fill(self, client_id, symbol, side, qty, price, ts): signed qty if side buy else -qty self.positions[symbol] self.positions.get(symbol, 0) signed self.cash - signed * price with conn() as c: c.execute(UPDATE orders SET statusfilled WHERE client_id?, (client_id,))client_id用「标的 时间戳 方向 数量」做哈希同样的参数在同一个 bar 上只会生成同一个 ID。第一次插入成功第二次撞主键直接返回不会重复成交。这就是幂等的核心让「重复提交」变成一个数据库层面的错误而不是靠代码里的 if 判断去猜。实盘实现live.py的差异在submit之后订单发给网关后进入pending回调里更新filled超时未回报的订单要主动查一次状态查到已成交就补记查到不存在就撤单重发。这套状态机建议单独抽一个OrderState枚举别用字符串硬编码。滑点和手续费在模拟盘里也要扣_apply_fill里把fill_price乘上一个滑点系数买入上浮、卖出下调比例放配置里常用万分之一到万分之三。回测里不扣这些实盘第一天就会给你上一课。5. 避坑与排查回测赚钱实盘亏钱的五个具体原因这套系统我前后改过几轮下面五条是踩得最疼、也最容易被忽略的。每一条都按「现象、原因、解决」写照着排查能省不少时间。5.1 信号用收盘价成交也用收盘价现象回测年化 30%模拟盘跑一个月只有 2%曲线形状倒是很像。原因回测引擎里策略出信号后直接按当前 bar 的close成交等于开了上帝视角——收盘价要等收盘才知道实盘里不可能在这个价格成交。解决成交价统一用「下一根 bar 的open」如果下一根不存在就顺延到再下一根。回测引擎里把broker.on_bar放在strategy.on_bar之前先撮合旧单再处理新信号。改完这一处年化通常会掉三分之一但掉下去的才是真实的。5.2 复权口径不一致指标全歪现象回测里某天的均线突然跳了一下信号跟着乱但原始行情看起来很正常。原因行情源默认返回不复权价格遇到分红送股时价格会断层另一路数据做了前复权两边拼在一起用均线自然断。更隐蔽的是「拉到一半换成复权」历史数据没重拉新旧混存。解决在normalize里显式加一列adj_type落库时带上回测启动前校验整段数据的adj_type是否一致不一致就整段重拉。复权方式选前复权还是后复权要固定别今天前复权明天后复权。5.3 网络超时后重试同一笔单下了两次现象模拟盘正常实盘偶尔出现双倍仓位查订单记录发现两笔参数完全一样的单。原因下单请求发出后超时程序以为是失败就重试实际上第一次已经成交了。没有幂等键的时候两次请求在网关眼里就是两笔独立订单。解决每笔单生成client_id并落库做主键重试时沿用同一个 ID网关侧如果不支持客户端订单号就在本地维护「pending 订单表」重试前先查状态。另外超时后不要立刻重发等一个查询周期确认未成交再重发。5.4 参数调到样本内最优样本外一塌糊涂现象双均线的fast/slow在回测区间里试了上百组最好的一组年化 45%、回撤 8%实盘跑两周就回撤 12%。原因在同一个区间反复试参数本质是在拟合这段行情的噪声。参数越多、试的次数越多过拟合越严重回测指标越好看越可疑。解决把数据切成训练段和验证段训练段调参、验证段只跑一次参数网格不要超过三维每组参数至少要求跨越两段不同行情最终上线前再跑一段完全没参与调参的数据指标衰减超过一半就放弃。5.5 日志只记结果出问题无法复盘现象某天权益突然少了一截翻日志只有一行「今日收益 -3%」不知道是哪笔单、哪个信号、什么价格造成的。原因日志只记了汇总指标没记每根 bar 的信号、每笔订单的请求参数和回报、每次风控拦截的原因。系统变成一个黑匣子出问题只能靠猜。解决用结构化日志每轮任务输出一行 JSON字段包括ts、symbol、signal、target_qty、client_id、fill_price、equity、risk_block。日志按天切文件保留至少三个月。风控拦截时把命中的是哪道闸也写进去排查时一眼就能看出是仓位上限还是熔断。提示日志里不要打印完整的账户号和密钥脱敏之后再落盘。排查需要的是订单号和参数不是凭证。6. 二次开发加策略、换通道、做验证的三步走拿到这套代码之后多数人要做的第一件事是把自己的策略塞进去。流程不复杂继承Strategy实现warmup、reset、on_bar三个方法在registry.py里注册一行配置里把strategy名字改掉。真正花时间的是验证不是编码。第一步是单测。给策略喂一段构造好的 bar 序列断言在指定位置应该出什么方向的信号。双均线这种简单策略手算三根均线的交叉点就能写断言复杂策略至少测「预热期不出信号」「连续同向信号只出一次」「数据不足时返回 None」这三条。回测引擎也值得测构造五根 bar验证成交价确实是下一根的开盘价。第二步是样本外验证。训练段调参验证段只跑一次两段的年化、回撤、夏普放在一张表里对比。我的习惯是要求验证段的年化不低于训练段的 50%回撤不超过 1.5 倍达不到就不上线。这一步没有捷径参数越少越容易通过。验证轮次数据范围通过标准目的样本内训练段年化、回撤可接受确认策略逻辑成立样本外验证段年化不低于训练段 50%排除过拟合模拟盘最近 30 个交易日信号与回测一致率 95% 以上验证工程链路小资金实盘 2 周滑点、成交率在预期内验证执行通道第三步换执行通道。execution/live.py里只要实现submit、position、equity、daily_pnl_pct四个方法其余代码不用动。换通道时最容易出问题的是「持仓同步」本地记的持仓和柜台不一致常见原因是手工下过单或者部分成交没回报。我的做法是每轮任务开始前先拉一次柜台持仓和本地比对不一致就以柜台为准并告警。最后一个技巧是给策略加一个「影子模式」新策略先不下单只把信号写进日志和现有策略的信号并排记录两周观察两者在同一根 bar 上的分歧点。分歧往往集中在开盘跳空和停牌复牌附近这些位置的信号质量最差提前发现能避免不少亏损。这套东西我从第一版到现在最大的教训是不要相信任何一次回测结果包括自己刚跑出来的。后来我每次上线新策略都强制走一遍「样本内、样本外、模拟盘、小资金」四轮任何一轮不达标就退回去改不给自己找理由跳过。希望这套骨架和这些坑能帮你在搭自己的 Python 算法交易系统时少走点弯路。本文还有配套的精品资源点击获取