简介：面向期望用 Python 从零搭建全自动量化交易系统的开发者，这份 38KB 的代码包覆盖从行情数据获取、技术指标计算、策略回测到交易执行与绩效统计的完整链路。压缩包共 35 个文件，其中 34 个为 Python 脚本，另附 1 个 Markdown 说明文档，脚本按功能模块拆分，包括 yfinance、Alpha Vantage、StockRow 等数据接口，RSI、ADX、MACD、布林带等指标计算，以及夏普比率、索提诺比率、最大回撤等风控绩效模块。策略示例包含突破策略、均值回归、神奇公式、Piotroski F 等多类量化模型，并附组合再平衡、波动率、斜率分析及可视化工具。这些代码大多可直接运行，有助于新手理解从原始数据清洗、指标构建到回测评估的完整流程，也方便有经验者快速迁移组件到自有交易系统。目前已有 3954 人学习，适合作为入门算法交易与量化选股的工程参考。
1. 把 Python 搬进交易终端：算法交易到底改变了什么
过去五年我见过太多人抱着“写个脚本自动赚钱”的念头冲进量化交易，又在第一周被数据源、回测偏差和实盘滑点劝退。Python 在这个领域之所以流行，不是因为语法优雅，而是因为它把“拿数据—算信号—下订单—看结果”这条链路压缩到了可以用一台普通电脑跑完的程度。这份资源讲的就是全自动交易系统：从行情获取、策略信号、回测验证到模拟盘执行的完整闭环。它适合已经会写基础 Python、想把自己的交易想法变成可验证程序的从业者，也适合刚入门但不想把时间浪费在造轮子上的新手。你要先明白一件事：自动化的核心不是省掉手动点击，而是让每一次买卖决策都能被历史数据检验、被风控规则约束。
2. 策略设计与数据准备：先让输入干净，再谈信号
2.1 数据源选型：日线起步，分钟线要留意复权与停牌
算法交易的第一步永远是数据，但这一步最容易翻车。我发现不少初学者直接下载某平台导出的 CSV 就开始算指标，结果除权除息日出现价格断层，信号被扭曲得一塌糊涂。常见的做法是优先选择后复权数据做回测，因为后复权保留了历史价格的真实收益关系，前复权数据在最新交易日附近会有价格偏移，不适合长周期回测。
import pandas as pd from pathlib import Path def load_klines(csv_path: Path) -> pd.DataFrame: """读取行情CSV，统一列为标准格式""" df = pd.read_csv(csv_path, parse_dates=["datetime"]) df = df.sort_values("datetime").drop_duplicates(subset="datetime") df = df.set_index("datetime") # 后复权处理通常由数据源完成，这里只做增量校验 df["ret"] = df["close"].pct_change() # 剔除极端跳变点，防止除权没处理干净混入信号 df.loc[abs(df["ret"]) > 0.2, ["open", "high", "low", "close"]] = pd.NA df = df.ffill().dropna() return df这段代码做了三件关键的事：按时间排序并去重、计算收益率列、把单日涨跌幅超过 20% 的可疑 K 线置空后向前填充。参数0.2是个经验阈值，同时适用于股票和多数数字资产，但如果你在跑期货数据，要按品种的涨跌停幅度调整，螺纹钢和股指的阈值必须分开设置。pd.NA配合ffill()是处理停牌日的常用组合，缺点是会把停牌日的开盘价和收盘价填充成前后交易日的数据，所以只适合做趋势类指标，不适合做开盘跳空类策略。
2.2 指标与信号：动量、均值回归和它们的组合边界
策略信号是整个系统的灵魂，但大多数入门资源只会告诉你“金叉买入、死叉卖出”，没有解释为什么这个逻辑在不同市场状态下表现会天差地别。动量类策略赚的是趋势延续的钱，适合单边行情；均值回归策略赚的是价格偏离修复的钱，适合震荡行情。把两者混在一个模型里不是不行，而是必须有明确的切换条件。
def add_signals(df: pd.DataFrame, fast: int = 10, slow: int = 30) -> pd.DataFrame: """计算快慢均线与通道突破信号""" df["ma_fast"] = df["close"].rolling(fast).mean() df["ma_slow"] = df["close"].rolling(slow).mean() df["ma_diff"] = df["ma_fast"] - df["ma_slow"] df["upper"] = df["close"].rolling(20).mean() + 2 * df["close"].rolling(20).std() df["lower"] = df["close"].rolling(20).mean() - 2 * df["close"].rolling(20).std() cond_cross_up = (df["ma_diff"] > 0) & (df["ma_diff"].shift(1) <= 0) cond_cross_down = (df["ma_diff"] < 0) & (df["ma_diff"].shift(1) >= 0) df["signal"] = 0 df.loc[cond_cross_up, "signal"] = 1 df.loc[cond_cross_down, "signal"] = -1 return df这里把均线交叉和布林带突破放在同一个函数里，实际使用时你只需要取其中一个条件。rolling(fast).mean()是核心计算逻辑，shift(1)用来判断“上一根 K 线还没满足、这一根才满足”的临界点，避免用未来数据。参数fast=10, slow=30只在日线级别表现稳定，切到 15 分钟线需要重做参数扫描。另一个细节是布林带的2 * std，这个 2 不是固定值，波动率大的品种可以放宽到 2.5，否则信号产生频率会低到策略根本无法运行。
2.3 从信号到订单：仓位、价格和触发时机的三角约束
信号只是告诉你“该动手了”，真正执行订单时还有三个绕不开的问题：买多少、什么价格买、什么条件放弃。仓位计算最常见的做法是固定比例法，即每次投入总资金的 2% 作为风险敞口，然后根据止损距离反推手数；也有用波动率倒数的，比如平均真实波幅越高仓位越小，适合趋势跟踪策略。
def build_orders(df: pd.DataFrame, capital: float = 100000.0, risk_ratio: float = 0.02, atr_n: int = 14) -> pd.DataFrame: df["atr"] = df["close"].rolling(atr_n).apply( lambda x: max(x.max() - x.min(), x[-1] - x[-2]), raw=True ) df["pos_size"] = 0 active = False for i in range(1, len(df)): if df["signal"].iloc[i] == 1 and not active: entry = df["close"].iloc[i] stop = entry - 2 * df["atr"].iloc[i] risk_per_unit = entry - stop df.loc[df.index[i], "pos_size"] = (capital * risk_ratio) / risk_per_unit active = True elif df["signal"].iloc[i] == -1 and active: df.loc[df.index[i], "pos_size"] = 0 active = False return df这段循环在逐行遍历，速度慢但逻辑直观，适合回测阶段验证信号组合的可行性。risk_ratio=0.02是经典的单笔风险上限，意思是这笔交易如果触发止损，亏损不超过总资金的 2%。pos_size的计算结果是标的数量，但在实际下单时还要整除合约乘数，比如股指期货一手对应一个固定面值，不能直接用这个结果下非整数单。止损用2 * atr是常见做法，数值越小止损越紧，信号触发频率越高但胜率越低，最终参数要通过后面的回测章节来确定。
3. 回测系统搭建：在真实资金进场前把逻辑跑透
3.1 用 Backtrader 搭最小回测框架
回测是算法交易里最不能省的一步，但我见过太多人打开 Jupyter Notebook 随便跑个矩阵就以为自己完事了。专业一点的套路是用 Backtrader 这类事件驱动框架，它能把订单撮合、佣金扣除、滑点模拟这些实盘要素一并纳入，而不是只算一个“理想收益率”。事件驱动回测的另一个好处是代码结构和你未来写实盘交易系统高度一致。
import backtrader as bt class MaCrossStrategy(bt.Strategy): params = (("fast", 10), ("slow", 30), ("risk", 0.02)) def __init__(self): self.fast_ma = bt.indicators.SMA(self.data.close, period=self.p.fast) self.slow_ma = bt.indicators.SMA(self.data.close, period=self.p.slow) self.cross = bt.indicators.CrossOver(self.fast_ma, self.slow_ma) self.order = None def next(self): if self.order: return if not self.position: if self.cross > 0: size = int((self.broker.getvalue() * self.p.risk) / self.data.close[0]) self.buy(size=size) elif self.cross < 0: self.close()CrossOver会在快线上穿慢线时返回 1、下穿时返回 -1，省去了手工判断shift(1)的步骤。size的计算用的是当前总资金乘以风险比例再除以现价，注意这里没有乘合约乘数，如果你在回测期货，需要额外处理。self.order的存在是为了防止同一根 K 线内重复下单，这是新手最容易踩的问题之一。有了策略类之后，还需要喂数据并运行：
from backtrader.feeds import PandasData def run_backtest(df: pd.DataFrame, cash: float = 100000.0): cerebro = bt.Cerebro() data_feed = PandasData(dataname=df, openinterest=None) cerebro.adddata(data_feed) cerebro.addstrategy(MaCrossStrategy) cerebro.broker.setcash(cash) cerebro.broker.setcommission(commission=0.0003) cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name="sharpe") cerebro.addanalyzer(bt.analyzers.DrawDown, _name="drawdown") result = cerebro.run() return result[0].analyzers.drawdown.get_analysis()注意setcommission(commission=0.0003)表示万三手续费，这是股票交易的常见费率，但如果你在做 ETF，手续费可能低到万 0.5；做期货则是按手数收费，要用setcommission(commission=3.0, mult=10.0)这样的形式单独设置。PandasData默认按索引做日期对齐，如果你的 DataFrame 没有把时间列设为索引，回测结果会错位到无法直视。
3.2 滑点与手续费：两种吞收益的隐形税
回测结果好不好看，很大程度取决于你对滑点和手续费的建模力度。滑点在回测里的标准做法是让成交价偏离信号价一个固定比例或一个最小跳动单位，比例设得过小，回测曲线漂亮但实盘对不上；设得过大，策略直接被摩擦成本拖平。
cerebro.broker.set_slippage_perc(perc=0.0005)set_slippage_perc表示每次成交在信号价格基础上向不利方向偏移 0.05%，这个值在流动性好的大盘股上够用，在小市值股票上可能要调到 0.15%。我一般会在回测时用两套参数对比：一套是理想值（滑点 0.01%），一套是保守值（滑点 0.1%），如果策略在保守值下依然盈利，才值得继续往下走。手续费和滑点加起来，单次买卖成本通常在 0.1%~0.2% 之间，一个年化 20% 的策略如果每月调仓 20 次，光是摩擦成本就吃掉一半收益，这是数学层面的硬约束。
3.3 评估指标：年化、回撤、夏普与盈亏比
回测跑完不能只看最终资金，要同时看四个指标：年化收益率、最大回撤、夏普比率和盈亏比。最大回撤决定了你的资金曲线能不能拿得住，回撤 30% 的策略即使年化很高，实盘时你大概率会在最低点附近手动止损退出。
def evaluate_metrics(result): sharpe = result[0].analyzers.sharpe.get_analysis().get("sharperatio", 0) drawdown = result[0].analyzers.drawdown.get_analysis() max_dd = drawdown.max.drawdown final_value = result[0].broker.getvalue() returns = (final_value / 100000.0) - 1 return {"年化": returns, "最大回撤": max_dd, "夏普": sharpe}夏普比率保守要大于 1.3 才能算及格，大于 2 说明策略逻辑大概率有过拟合嫌疑，需要警惕。最大回撤建议控制在 15% 以内，超过这个数字，你要么降低仓位，要么缩减止损距离。评估指标的意义不是找一个“最好看”的数字组合，而是检验策略在你愿意承受的心理压力范围内是否稳定。
4. 避坑指南：回测曲线漂亮，实盘翻车的六个真实原因
4.1 未来函数：信号里混进了未来数据
现象：回测收益曲线近乎 45 度角向上，最大回撤不到 3%，但拉到实盘模拟盘里连续亏损。 原因：最常见的是在计算信号时用了close的当前值去判断当天收盘是否突破，同时又把这个信号用在当天开盘价上，等于用收盘价预测开盘价。 解决：在信号生成时统一用shift(1)滞后一根 K 线，确保第 N 根 K 线的信号是基于第 N-1 根及之前的数据。代码审查时逐行问一句“这行代码在当根 K 线收盘前能不能拿到数据”，拿不到就是未来函数。
4.2 回测撮合逻辑太理想
现象：回测每次都按收盘价完美成交，实盘时买入价格总是比信号价高一点，卖出总是低一点。 原因：回测里没有设置滑点和最小变动价位，特别是市价单被简化成了“以当前 Bar 收盘价成交”。 解决：统一设置set_slippage_perc(0.0005)，并把最小交易量调成一手，观察收益曲线是否还能站上基准。如果保守参数下曲线依然稳定，再谈实盘。
4.3 手续费参数失真
现象：策略持仓周期只有一两天，回测盈利但扣除手续费后变成负收益。 原因：把手续费设成了 0 或者万 0.5，没有按实际券商费率计入。 解决：按你真实开户券商的费率填写，股票交易还要加上卖出时的印花税。频率越高的策略对费率越敏感，持仓周期小于 3 天的策略，手续费费率差 0.1% 就可能决定盈亏。
4.4 参数过拟合
现象：做参数扫描时发现(fast=8, slow=21)的夏普高达 3.8，但换一段时间的样本外数据立刻失效。 原因：策略在历史数据里找到了一个恰好匹配的噪声组合，换个时间段就失效。 解决：把样本数据切分为训练集和验证集，训练集做参数扫描，验证集只做一次验证。如果验证集结果不如训练集的一半，这个参数组合就是过拟合信号。
4.5 停牌与涨跌停未处理
现象：回测在某只股票上产生了多次买入信号，但实盘根本买不进。 原因：回测数据里没有标记停牌日和涨跌停限制，信号在涨停板上触发买入。 解决：在数据预处理阶段剔除停牌日，并对涨停和跌停日的买卖信号做屏蔽。我的做法是额外读取一张交易日历表，把非交易日和涨跌停板都打上标记，再在策略里过滤。
4.6 资金管理缺失
现象：策略单笔收益不错，但连续亏损后资金曲线跌到起跑线以下。 原因：每次交易都是满仓进出，没有按固定风险比例分配仓位。 解决：用第 2.3 节的risk_ratio方法，把单笔风险控制在 2% 以内。资金管理的意义不是提高收益，而是保证你在策略表现不佳的时期依然能活到盈利到来。
5. 自动化部署：从回测脚本到无人值守的交易终端
5.1 定时任务：APScheduler 与系统 cron 的取舍
策略跑通了之后，下一步是让它每天自动运行，而不是你手动打开电脑点击“运行”。我在实际项目中用到的最稳妥方案是 Linux 服务器的crontab配合 Python 脚本日志输出，简单直接；如果你的交易时间窗口需要精细控制，比如开盘前 15 分钟拉数据、开盘后 5 分钟下单，则用 APScheduler 更灵活。
from apscheduler.schedulers.blocking import BlockingScheduler from datetime import datetime def job_fetch_data(): print(f"[{datetime.now()}] 开始拉取行情") # 实际场景这里调用数据源API并落库 def job_run_strategy(): print(f"[{datetime.now()}] 执行策略信号计算") # 读取最新数据，计算信号，生成目标持仓 scheduler = BlockingScheduler(timezone="Asia/Shanghai") scheduler.add_job(job_fetch_data, "cron", hour=9, minute=15) scheduler.add_job(job_run_strategy, "cron", hour=9, minute=20) scheduler.start()BlockingScheduler会阻塞主进程，适合专用服务器；如果你有其他任务要同时跑，改用BackgroundScheduler。timezone参数务必显式指定，服务器默认时区可能是 UTC，导致 9 点 15 分的任务在下午 5 点才触发。所有任务函数内部要包一层try...except，把异常写入日志文件，否则脚本崩溃后没有任何痕迹可查。
5.2 交易状态机：处理部分成交与撤单重试
模拟盘和实盘的最大差异在于订单状态。回测里你下一张单，要么全成交要么不成交；真实环境里，大单会被拆成多笔在不同价位成交，甚至有一半成交、一半挂着的状态。写自动交易系统必须维护一个订单状态机，不能下完单就不管了。
class OrderState: PENDING = "pending" PARTIAL = "partial" FILLED = "filled" CANCELED = "canceled" class TradeExecutor: def __init__(self, broker_api, max_retry: int = 3): self.api = broker_api self.max_retry = max_retry def place_order(self, symbol: str, side: str, amount: float): resp = self.api.create_order(symbol, side, amount) if resp["status"] == OrderState.PENDING: for _ in range(self.max_retry): status = self.api.query_order(resp["order_id"]) if status["status"] == OrderState.FILLED: return status if status["status"] == OrderState.PARTIAL: remaining = status["remaining"] self.api.cancel_order(resp["order_id"]) self.api.create_order(symbol, side, remaining) break return respmax_retry=3是控制暴走的保险丝，防止订单在异常行情下反复提交。部分成交后的处理策略有两种：一种是取消剩余订单然后重新下一张更小的单，另一种是接受部分持仓，下一次信号来了再补。我倾向第二种，因为反复撤单会拉黑你的账户权限，某些券商的接口限制一天最多撤单 10 次，超了会被封禁到第二天。你还要在状态机里加入“订单超时未成交就主动撤回”的逻辑。
5.3 风控模块：止损、仓位上限与下单冷却
自动化系统里必须有一个独立于策略模块的检查层，策略负责赚钱，风控负责让你活下去。风控规则越简单越可靠，我见过有人写了 20 条风控规则，结果实盘时有一条规则触发了误判，把正常仓位全平了，损失比不设风控还大。
class RiskManager: def __init__(self, max_position: float = 0.3, max_daily_loss: float = 0.03, cooldown_seconds: int = 60): self.max_position = max_position self.max_daily_loss = max_daily_loss self.cooldown_seconds = cooldown_seconds self.last_trade_time = None def can_trade(self, current_position, daily_return): if current_position > self.max_position: return False if daily_return < -self.max_daily_loss: return False if self.last_trade_time: elapsed = (datetime.now() - self.last_trade_time).total_seconds() if elapsed < self.cooldown_seconds: return False return Truemax_position=0.3表示单票持仓不超过总资金的 30%，这是我做股票策略时的常用上限，防止单只票的黑天鹅拖垮净值。max_daily_loss=0.03是每日熔断线，亏到 3% 就停止当天所有交易，这比单独设止损更能保护账户。cooldown_seconds=60是下单冷却时间，防止策略在震荡行情里频繁触发买卖，把手续费吃穿。
5.4 日志与告警：无人值守时，脚本必须能“自己说话”
无人值守意味着没有人在屏幕前盯着终端，日志和告警就是你唯一的眼睛。我见过太多人部署完脚本后一周没看，登录服务器发现第二天就崩了，之后的几天都在亏钱空转。因此日志至少要记录三个维度：每次下单的完整请求与响应、每次风控拦截的理由、每次策略报错时的回滚状态。
import logging logging.basicConfig( filename="trade.log", level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s" ) def log_order(order_info): logging.info(f"ORDER {order_info}") # 关键字段：时间、标的、方向、数量、价格、状态、剩余量 def log_risk_block(reason): logging.warning(f"RISK_BLOCK {reason}")告警方面，我的做法是在风控触发和订单异常时调用一个简单的 webhook 通知：发送到手机通知应用。没必要搞复杂的告警平台，一条消息能让你在 5 分钟内知道系统出问题就够了。日志文件还要做轮转，logging.handlers.RotatingFileHandler可以控制单文件最大大小和保留份数，避免服务器磁盘被日志塞满。
6. 参数稳健性检验：从回测到模拟盘的最后一公里
策略在历史数据上表现好，不代表未来也表现好，这一步的核心工作是检验参数稳健性而不是追求收益最大化。我常用的方法是网格扫描加样本外验证。继续用双均线策略做例子，把fast从 5 扫描到 30，slow从 20 扫描到 60，记录每个参数组合在训练集上的夏普比率，然后看这些组合在验证集上的表现。
def grid_search(df_train, df_valid): results = [] for fast in range(5, 31, 5): for slow in range(20, 61, 10): if fast >= slow: continue # 训练集回测 train_sharpe = run_backtest(df_train, fast, slow) # 验证集回测 valid_sharpe = run_backtest(df_valid, fast, slow) results.append((fast, slow, train_sharpe, valid_sharpe)) return results观察结果时不是找验证集上最高的那个点，而是找“训练集和验证集表现都稳定在中等偏上”的区域。如果一个参数组合在验证集上的夏普从 2.5 掉到 0.4，说明该区域不稳定；如果训练集和验证集都能保持在 1.2 到 1.8 之间，这个区域更可能在未来延续。另一个常用检验是随机化检验：把收益率序列随机打乱 100 次，用同一套策略逻辑重新跑，如果打乱后的收益分布也出现了接近策略原本的成绩，说明策略的收益来源是噪声而不是真实规律。参数扫描的结果不应该被当作“最佳参数”锁死，我更倾向于把一组参数直接固化在中等水平，然后用加仓减仓规则去适应市场变化，而不是频繁调整参数。
模拟盘是最后一重验证，建议至少跑 4 到 6 周，跨越至少一次完整的行情起伏。模拟盘和回测的差距如果超过 15%，先别急着改参数，回去检查滑点设置和订单状态处理；如果模拟盘表现和回测基本一致，再考虑用小资金实盘验证。我从那以后每次把策略推向实盘，都强制走一遍网格扫描、样本外验证、模拟盘对照这三个流程，缺一不可。参数稳健性检验可能让策略少赚 10% 的纸面收益，却能避免你在实盘里亏掉真正重要的东西。希望帮到你。
本文还有配套的精品资源，点击获取
