简介：一套面向金融量化初学者及具备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

依赖不多，核心就几个。版本不用追新，能锁就锁，回测结果对不上时第一步就是核对依赖版本。

行情拉取那一层建议做成可替换的适配器：定义一个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"], required=True) p.add_argument("--config", default="config/config.yaml") p.add_argument("--strategy", default="dual_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, broker=PaperBroker(ctx.cfg)) else: from app.scheduler import run_live run_live(ctx, broker=LiveBroker(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(minutes=FREQ_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, cash=100_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( signal=signal, equity=self.broker.equity(price), price=price, current_qty=self.broker.position(signal.symbol), ) self.broker.submit(ts, signal.symbol, target, bar, reason=signal.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里periods=252对应日频，如果回测用的是小时线，要改成对应周期数，否则夏普会虚高。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, f"MA{fast_n}>{slow_n}") if fast < slow and self.holding: self.holding = False return Signal(bar["symbol"], "sell", 1.0, f"MA{fast_n}<{slow_n}") return None

history只追加不裁剪在长回测里会吃内存，实际项目里可以改成固定长度的deque(maxlen=slow_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 status='filled' 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 倍，达不到就不上线。这一步没有捷径，参数越少越容易通过。

第三步换执行通道。execution/live.py里只要实现submit、position、equity、daily_pnl_pct四个方法，其余代码不用动。换通道时最容易出问题的是「持仓同步」：本地记的持仓和柜台不一致，常见原因是手工下过单或者部分成交没回报。我的做法是每轮任务开始前先拉一次柜台持仓，和本地比对，不一致就以柜台为准并告警。

最后一个技巧是给策略加一个「影子模式」：新策略先不下单，只把信号写进日志，和现有策略的信号并排记录两周，观察两者在同一根 bar 上的分歧点。分歧往往集中在开盘跳空和停牌复牌附近，这些位置的信号质量最差，提前发现能避免不少亏损。

这套东西我从第一版到现在，最大的教训是：不要相信任何一次回测结果，包括自己刚跑出来的。后来我每次上线新策略，都强制走一遍「样本内、样本外、模拟盘、小资金」四轮，任何一轮不达标就退回去改，不给自己找理由跳过。希望这套骨架和这些坑，能帮你在搭自己的 Python 算法交易系统时少走点弯路。

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