简介一份Python基于机器学习的量化投资策略实战项目面向金融量化初学者与数据分析人员解决从行情数据获取、特征工程到模型训练与回测评估的全流程问题。压缩包共15个文件约735KB含5个Python脚本、6张png图表、2个txt股票代码列表与依赖包清单、README使用说明及.gitignore目录按数据、特征、模型等模块分隔便于定位修改。目前已有267人学习下载。项目以10支股票为demo通过数据接口拉取历史价格构造高级特征并保存为pickle使用LightGBM训练选股模型再用回测脚本统计累积收益、最大回撤、夏普率三项核心指标。配套说明清晰给出分步执行与一键运行两种方式可帮助读者快速理解各文件职责并将其扩展到更大的股票池是一份适合动手实践的量化入门参考能直观看到从原始行情到交易指标的完整逻辑。1. 用机器学习做量化投资回测能过实盘能活吗如果你手里有一套基于 Python 的机器学习量化投资策略源码最要紧的事不是急着看收益曲线而是先回答一个问题回测里的收益到底是策略本事还是数据泄漏的幻觉。我的经验是用 Python 跑一套“机器学习预测涨跌 回测验证”的流程不难真正难的是让模型在样本外、扣掉手续费滑点之后依然稳定。这个方向适合两类人一类是已经写过均线、MACD 这类规则策略想升级到机器学习模型另一类是有 Python 基础想系统了解量化策略从数据到回测的完整链路。本文会给你一套可复现的最小骨架——从数据清洗、特征工程、LightGBM 训练到 Backtrader 回测再把最常见的几个坑逐个拆开。2. 特征工程与数据对齐先解决“喂给模型什么”再谈预测准不准2.1 行情数据对齐用前复权价格按日期拼成一张表机器学习模型吃的是一张二维表行是交易日列是特征。第一步就是把原始行情拉下来统一列名、统一索引、统一复权方式。常见做法是直接用 akshare 或 tushare 拉 A 股日线我一般用 akshare 的前复权数据做历史回测原因是前复权价格能反映实际可交易的连续价格序列不会因为除权除息产生跳空假信号。import pandas as pd import akshare as ak def load_daily_bar(symbol: str, start: str, end: str) - pd.DataFrame: # 拉取前复权日线数据 raw ak.stock_zh_a_hist( symbolsymbol, perioddaily, start_datestart, end_dateend, adjustqfq # qfq前复权hfq后复权 ) raw.columns [ date, open, close, high, low, volume, amount, amplitude, pct_chg, change, turnover ] df raw[[date, open, high, low, close, volume]].copy() df[date] pd.to_datetime(df[date]) df.set_index(date, inplaceTrue) df df.astype(float) return df # 示例拉取贵州茅台 2020-2024 日线 # df load_daily_bar(600519, 20200101, 20241231)这里有两个容易翻车的点。第一akshare 返回的列名是中文如果直接用原始列名后面pct_change()之类的操作没问题但合并多只股票时容易乱所以我先统一成英文列名。第二日期要转成datetime并设为索引否则后续rolling和shift的时间对齐会出问题——这是新手最容易忽略的步骤。2.2 特征构建动量、波动率、量价关系全靠 shift 防未来函数特征工程是整个机器学习量化流程里最值得花时间的部分。常见做法是构造三类特征动量类过去 N 日涨幅、波动类收益率的滚动标准差、量价类成交量与价格的背离关系。但有个铁律任何特征都不能用到“当天收盘以后才知道的数据”所以只能用shift()和rolling()的历史窗口。import numpy as np def build_features(df: pd.DataFrame, lookback: int 10) - pd.DataFrame: df df.copy() # 基础收益当日收盘相对昨日收盘 df[ret_1] df[close].pct_change() # 动量特征过去 lookback 日累计涨幅 df[mom] df[close] / df[close].shift(lookback) - 1 # 波动特征过去 20 日收益标准差衡量波动率 df[volatility] df[ret_1].rolling(20).std() # 均线偏离收盘价相对 20 日均线的偏离度 df[ma_20] df[close].rolling(20).mean() df[ma_bias] df[close] / df[ma_20] - 1 # 量能变化当日成交量相对 20 日均量的倍数 df[volume_ratio] df[volume] / df[volume].rolling(20).mean() # 量价相关性过去 5 日收益率与成交量变化的相关系数 def rolling_corr(a, b, window5): return a.rolling(window).corr(b) df[ret_volume_corr] rolling_corr(df[ret_1], df[volume].pct_change()) # 去掉无穷值和缺失值否则模型训练直接报错 df df.replace([np.inf, -np.inf], np.nan) df df.dropna() return df参数说明lookback控制的是动量窗口这里取 10 个交易日约等于两周volatility和ma_20固定用 20 日是 A 股日频策略里比较常见的周期。如果你想做得更细致可以把这个函数改造成接收多个 lookback 参数然后拼出一组特征比如 5 日、10 日、20 日的动量都放进同一张表。但要注意特征越多过拟合风险越高后面要配合正则化参数一起交代。2.3 多股票数据合并不要把所有股票直接堆进一张表多股回测的热门做法是把多只股票的日线数据纵向拼接然后训练一个模型。这样做不是不行但你要清楚一个前提不同股票的波动水平差异很大茅台和一只小盘股的收益率分布完全不是一回事。常见做法是每只股票单独做特征再按日期横向合并或者干脆逐股训练独立模型。下面是横向合并的模板def merge_multi_stocks(symbols, start, end): panels {} for sym in symbols: price load_daily_bar(sym, start, end) feature build_features(price) # 给列名加上股票代码前缀避免同名特征冲突 feature.columns [f{sym}_{c} for c in feature.columns] panels[sym] feature # 按日期索引做外连接缺失值先保留后面再统一处理 merged pd.concat(panels, axis1) return merged这样合并出来的表每一行代表一个交易日每一列代表某只股票的某个特征。模型输入维度大了很多但好处是可以通过列名回溯到具体股票的特征排错和解释都比较直观。这里有个取舍横向合并要求所有股票上市时间重叠否则早期大量缺失值会直接丢掉样本所以样本池需要先过滤一遍数据的起始日期。3. 模型训练与评估LightGBM 做日频方向预测参数怎么调才不过拟合3.1 标签定义预测未来 5 日涨跌方向而不是预测收益率数值量化策略里的机器学习通常是把预测目标定义为“未来 N 日收益的符号”。为什么不直接回归预测收益率数值因为日频收益率信噪比极低回归模型很难稳定拟合而二分类只看方向和概率置信度模型更容易学习到有区分度的模式。这里我采用 N5也就是一周左右的持有期符合常见的短线策略节奏。def make_labels(df: pd.DataFrame, horizon: int 5): df df.copy() # 未来 horizon 日收益率用 shift(-horizon) 滚动窗口求未来的收益 df[future_ret] df[close].shift(-horizon) / df[close] - 1 # 标签未来收益 0 记为 1否则为 0 df[label] (df[future_ret] 0).astype(int) # 删除最后 horizon 行因为这些行没有未来数据label 为 NaN df df.dropna(subset[label]) return df逻辑说明shift(-horizon)是“未来函数”的合法使用场景——因为我们是在构造标签不是在构造特征。模型训练时输入的 X 不能包含未来数据但标签 Y 本身就必须指向未来否则这个策略就没有预测目标。注意最末尾的 N 行样本因为拿不到未来收益必须直接删掉否则会把 NaN 标签喂给模型。3.2 时序交叉验证为什么不能用随机划分以及 TimeSeriesSplit 的用法很多新手拿到数据后习惯用train_test_split(test_size0.2, random_state42)随机切分。这在普通机器学习任务里没问题但时序数据一旦随机打乱就相当于允许模型“偷看”未来行情来预测过去回测指标会高得离谱。正确的做法是严格按时间顺序切分训练集永远在测试集之前。from sklearn.model_selection import TimeSeriesSplit from sklearn.metrics import accuracy_score, roc_auc_score import lightgbm as lgb def train_lgb_model(df, feature_cols): df df.sort_index() # 确保按时间升序 X df[feature_cols].values y df[label].values # 5 折时序切分训练集是第 1 段验证集是第 2 段以此类推 tscv TimeSeriesSplit(n_splits5) oof_pred np.zeros(len(X)) for train_idx, valid_idx in tscv.split(X): X_tr, X_va X[train_idx], X[valid_idx] y_tr, y_va y[train_idx], y[valid_idx] model lgb.LGBMClassifier( n_estimators200, learning_rate0.05, num_leaves7, # 小叶子节点数防止过拟合 max_depth3, feature_fraction0.8, bagging_fraction0.8, bagging_freq1, random_state42, verbosity-1 ) model.fit( X_tr, y_tr, eval_set[(X_va, y_va)], callbacks[lgb.early_stopping(stopping_rounds50)] ) oof_pred[valid_idx] model.predict_proba(X_va)[:, 1] # 用 OOF 预测计算整体 AUC衡量模型排序能力 auc roc_auc_score(y, oof_pred) print(fOOF AUC: {auc:.4f}) return model参数说明num_leaves7和max_depth3是刻意调小的这是时间序列预测里非常关键的一步——模型越复杂越容易记住特定行情的噪声样本外一换行情就失效。feature_fraction和bagging_fraction是随机采样参数类似随机森林的列采样和行采样实战中一般保持在 0.8 左右太低会影响模型容量。early_stopping在验证集上训练轮数超过 50 没提升就停止既省时间又避免过拟合。3.3 模型评估准确率不是最重要的IC 和分层单调性才是模型训练完之后很多人只看准确率如果准确率 52% 就觉得可以了。但二分类准确率在涨跌标签不平衡时没有意义——如果 55% 的日子在涨一个“永远预测涨”的傻瓜模型准确率就是 55%。正确的评估方式是看预测概率与未来收益的排序相关性也就是 ICInformation Coefficient。from scipy.stats import spearmanr def evaluate_ic(df, pred_prob): # 预测概率与真实未来收益的秩相关衡量排序能力 ic, pvalue spearmanr(pred_prob, df[future_ret]) print(fIC: {ic:.4f}, p-value: {pvalue:.4f})我自己的经验是日频策略的 IC 能在 0.03 到 0.05 之间就算可用0.01 以下的策略基本就是随机噪声。除了 IC 之外还可以把预测概率按十分位数分组看 top 组的平均收益是否显著高于 bottom 组如果分位数收益单调递增说明模型的排序能力是真实的如果不单调说明模型只是在个别样本上碰巧预测对了这种策略到实盘大概率会翻车。4. 用 Backtrader 跑通 ML 策略回测从预测信号到净值曲线的完整代码4.1 信号与成交的时序约束T 日收盘生成信号T1 开盘才允许成交模型预测完之后最容易被忽略的是信号的使用时机。如果 T 日收盘后模型生成“明天上涨”的预测然后你在回测里直接用 T 日收盘价买入这就是前视偏差——实盘里你根本不可能在生成信号的同一点位成交。正确做法是T 日收盘后计算特征并预测T1 日开盘价成交。这个约束直接决定了回测的可靠程度。4.2 写一个最小可用的 Backtrader 策略类Backtrader 是 Python 生态里最常用的回测框架。核心是把预测信号预先算好作为一个外部 DataFrame 传入策略然后在next()里通过当前日期去查信号表而不用在策略内部重算模型——这样代码更清晰也便于排查信号对齐问题。import backtrader as bt class MLSignalStrategy(bt.Strategy): params dict( prob_threshold0.55, # 预测概率大于 0.55 才买入 hold_days5 # 最长持有一天数超过则强制卖出 ) def __init__(self, signal_df): # signal_df: DataFrame索引是日期包含 prob 列取值为 0~1 self.signal_df signal_df self.hold_count 0 self.order None def next(self): if self.order: return # 有未完成订单时跳过本周期 current_date self.data.datetime.date(0) # 从预计算的信号表里取当日信号 if current_date in self.signal_df.index: prob self.signal_df.loc[current_date, prob] else: prob 0.0 # 没有持仓时概率超阈值则买入 if not self.position: if prob self.params.prob_threshold: self.order self.buy() self.hold_count 0 else: self.hold_count 1 # 持有超过 hold_days 或者信号转弱卖出 if self.hold_count self.params.hold_days or prob 0.5: self.order self.close()逻辑说明策略里signal_df是外部传入的预测结果表current_date是当前 bar 的日期用这个日期去查表。这一步有几个细节要注意self.data.datetime.date(0)返回的是datetime.date对象而signal_df.index如果是datetime64直接判断in可能失效最稳妥的做法是把signal_df.index转成pd.DatetimeIndex后用.date统一格式。参数说明prob_threshold0.55表示只有模型给出 55% 以上上涨概率才开仓可以有效过滤低置信度信号hold_days5是强制离场周期防止某个信号“套牢”后无限持有。这两个参数都需要在后续的敏感性分析里反复调整。4.3 回测引擎设置现金、佣金、滑点一个都不能省回测引擎的参数直接决定了结果是否可信。我的习惯是最少设置三样初始资金、手续费、滑点。A 股佣金按万 2.5 比较常见卖出时还要考虑印花税但 Backtrader 的setcommission只能设一个统一比例所以我会用 0.0003 覆盖买卖双侧成本滑点按 0.001千分之一起步这代表每次成交都比理想价格多付出 0.1% 的成本。cerebro bt.Cerebro() cerebro.addstrategy(MLSignalStrategy, signal_dfsignal_df) # 把数据喂给 cerebro data_feed bt.feeds.PandasData(datanameprice_df) cerebro.adddata(data_feed) # 账户与成本设置 cerebro.broker.setcash(1_000_000.0) # 初始资金 100 万 cerebro.broker.setcommission(commission0.0003) # 万 3 综合费率 cerebro.broker.set_slippage_perc(perc0.001) # 千分之一滑点 # 添加绩效分析器 cerebro.addanalyzer(bt.analyzers.SharpeRatio, _namesharpe, riskfreerate0.0, annualizeTrue) cerebro.addanalyzer(bt.analyzers.DrawDown, _namedrawdown) cerebro.addanalyzer(bt.analyzers.Returns, _namereturns) results cerebro.run() strategy results[0] print(夏普比率:, strategy.analyzers.sharpe.get_analysis()[sharperatio]) print(最大回撤:, strategy.analyzers.drawdown.get_analysis()[max][drawdown]) print(年化收益:, strategy.analyzers.returns.get_analysis()[rnorm100])参数说明SharpeRatio 分析器里annualizeTrue表示按年化计算riskfreerate0.0是把无风险利率设为 0你可以按实际情况改成 0.02 或 0.03。set_slippage_perc的perc是比例滑点Backtrader 会在成交价上额外加上或扣掉这个比例用来模拟冲击成本。这里有一个容易看错的地方Backtrader 的setcommission(commission0.0003)传的是比例不是固定金额如果传commission3.0就变成每笔固定 3 元含义完全不同。4.4 回测结果解读单看总收益是新手行为要拆开看每笔交易跑完回测后除了夏普和回撤我还会把每笔交易记录导出来检查有没有异常交易——比如持仓时间只有 1 天、买入后立刻亏损卖出、成交价格异常偏离当日 K 线。一个能上实盘的策略交易记录应该是符合逻辑的开仓理由清晰、止损纪律明确、交易频率稳定。如果回测出来的策略一年交易 300 次每次收益都在 0.5% 附近那大概率是过拟合到噪声上实盘里不可能稳定复制。5. 五大回测陷阱排查账面收益和实盘收益之间到底差在哪5.1 未来函数回测曲线平滑得像教科书实盘却一塌糊涂现象回测净值曲线几乎不回撤年化收益高得离谱而且每个月的收益都稳定为正。原因十有八九是数据泄漏——可能是特征里用了未来数据也可能是信号在当天收盘价成交。解决我在代码里强制约束特征只用shift()后的历史数据成交价用次日开盘价。排查时可以把打印日志打开随机抽 10 笔交易人工核对这些交易的开仓时间点是否真的能拿到对应信号。5.2 训练集和测试集重叠模型其实在“开卷考试”现象回测 AUC 高达 0.75 以上样本外却跌回 0.5。原因构建特征时用了滚动窗口而标签也是滚动窗口如果切分时没有在训练集和验证集之间留出空隙相邻日期的样本会高度相关模型等于记住了测试集答案。解决在TimeSeriesSplit的基础上训练集与验证集之间留出至少一个horizon比如 5 个交易日的间隔让验证集完全不在训练窗口的“视野”内。实现上可以手动切分索引而不是全依赖TimeSeriesSplit默认参数。5.3 幸存者偏差你的股票池里全是活到今天的股票现象回测选了当前沪深 300 成分股回测区间拉长到 10 年结果收益很好。原因十年前还在沪深 300 里、后来退市或长期停牌的股票被你过滤掉了模型天然只见过“活得好的股票”。解决股票池要按回测当时的成分股快照来选或者至少纳入已经退市的股票。如果数据源拿不到退市股那么回测结果要打折看并且要重点观察策略是否依赖少数暴涨股票。5.4 过拟合网格搜索结果漂亮换个时间段立刻失效现象模型参数经过网格搜索后训练集 AUC 0.72回测收益年化 30%但把回测区间换成最近两年收益变成负的。原因参数被优化到特定行情上本质上是把噪声当成了信号。解决一是限制模型复杂度num_leaves不要超过 15二是做样本外验证把数据按时间分成三段训练调参段、验证段、最终测试段调参只看训练和验证最终测试段只在最后跑一次绝不回头调整。我自己的规则是如果模型参数需要反复试超过 20 组才能得到好结果说明特征本身没有预测力换特征比调参更重要。5.5 涨跌停和停牌回测显示能成交实盘根本排不上队现象回测里策略在涨停板上买入成功第二天卖出获利实盘却连续几天买不进。原因回测引擎假设任何价格都能成交但 A 股有涨跌停限制涨停时买单无法成交跌停时卖单无法成交。解决在回测里加一个过滤条件当信号触发当天是一字涨停就直接跳过这笔信号更严格的做法是检查信号触发当日的成交量是否足够大如果成交量过低说明流动性不足同样跳过。这个过滤在 Backtrader 里可以用自定义Order逻辑实现前期先做一个简化版把 10 日内发生过一字涨停的样本直接从测试集里剔除。6. 回测之后滚动训练与样本外验证把策略寿命拉长策略从回测到实盘之间最后一道工序是滚动训练验证也叫 Walk-Forward 分析。思路很简单每隔一段时间用最新数据重新训练模型然后用接下来的 20 个交易日做样本外预测把每段的样本外收益拼接起来得到一条完整的“真实绩效曲线”。这样能模拟实盘中你每隔一个月重新训练模型的过程避免一次性训练完成后行情风格变了还硬扛。def walk_forward_validate(df, feature_cols, train_days120, test_days20): 滚动训练验证 每 120 个交易日训练一次模型 预测接下来 20 个交易日拼接样本外信号。 df df.sort_index() prob_series pd.Series(indexdf.index, dtypefloat) for start in range(0, len(df) - train_days - test_days, test_days): train df.iloc[start:start train_days] test df.iloc[start train_days:start train_days test_days] model lgb.LGBMClassifier( n_estimators100, learning_rate0.05, num_leaves7, max_depth3, random_state42, verbosity-1 ) model.fit(train[feature_cols], train[label]) prob_series.iloc[start train_days:start train_days test_days] \ model.predict_proba(test[feature_cols])[:, 1] return prob_series.dropna()逻辑说明每次只预测未来 20 个交易日然后模型作废重新训练。这样做既避免了模型长期不更新导致失效又让样本外测试的规模足够大统计结论更可靠。参数说明train_days120约等于半年交易日适合日频策略test_days20约等于一个月这个节奏比较适合人工介入。滚动训练的缺点是一次参数调整要跑很多次模型时间成本高但这是机器学习量化策略上线前最值得投入的验证步骤。做完以上验证后我还有一个强制习惯把回测的滑点从 0.001 提高到 0.003、手续费从万 3 提到千 1重新跑一遍整条流水线。如果策略在翻倍的成本假设下还能盈利才会考虑用真金白银测试。如果一提高成本收益就直接变负那说明策略的真实超额收益很薄即使上线也只适合用小资金跑。量化策略的坑绝大多数不在数学而在细节。数据泄漏、成交假设、成本模型每一项都能把回测结果从天堂打回地狱。把刚才这些验证跳过任何一步都等于在黑匣子里开车看着仪表盘漂亮实际方向盘是失灵的。我自己就曾经拿一份 AUC 0.72 的策略上模拟盘结果三个月跑下来亏损 8%排查后才发现是训练验证集重叠导致的虚高回测。之后我强制自己把每一步都固定成模板验收不合格就不上线。希望帮到你少走这一段弯路。本文还有配套的精品资源点击获取