简介：这是一套面向量化投资学习者和金融科技从业者的Python机器学习实战资源，以基本面量化投资为核心场景，集成了线性回归、决策树、随机森林、支持向量机、XGBoost、LSTM等多种算法，覆盖数据预处理、特征工程、模型训练、策略回测与结果分析完整流程。资源包共219个文件，压缩后约145.79MB，其中167个CSV数据文件提供实验所需的完整市场与基本面数据，36个.abak文件为各算法的中间结果或模型备份，11个.py脚本为可直接运行的源码，另有PDF与Markdown文档辅助理解项目设计。源码已在本地编译并验证可运行，评审分达95分以上，项目难度适中，非常适合希望将机器学习理论落地到真实金融场景的初学者，也适合有一定基础的研究者参考算法选择与优化思路。目前已有49人学习下载，资源包含可直接复用的算法模型、实验数据集及配套说明文档，能够帮助用户快速搭建量化研究框架，并通过实际案例理解不同机器学习算法在选股、风控和策略优化中的应用逻辑，是一份兼具教学与实践价值的完整参考资料。
1. 用 Python 跑通机器学习量化投资研究：源码、数据集和多种算法从哪里下手
标题里的“Python实现机器学习驱动的量化投资研究项目源码及数据集，集成多种算法”，我把它翻译成一句话：一套从行情数据输入到多模型预测、再到回测评估的完整工程骨架，而不是某个单点算法。很多人在这个方向上翻车，不是因为模型不够高级，而是数据切分、标签构造和回测口径在互相打架。这篇笔记要解决的是“拿到这类项目后怎么看得懂、跑得动、改得动”，顺序是数据 → 标签 → 算法集成 → 回测 → 排坑。适合已经会写 pandas 和 sklearn 基础代码、但对量化研究流程还缺一张地图的从业者。
2. 数据与标签是量化机器学习的地基：先把 dataset 目录和 label 口径定下来
一个量化 ML 项目拿到手，我最先做的一件事不是运行main.py，而是打开数据集目录和构造标签的函数。因为模型再花哨，数据口径错了，后面全是白费。接下来把数据组织、标签定义、时间切分这三个决定研究成败的环节一次说清楚。
2.1 数据文件怎么组织：一券一表还是合并宽表
源码包里常见的数据集组织方式是“一券一表”：每只股票一个 CSV，字段至少包含日期、开盘、收盘、最高、最低、成交量。这样的好处是单标的清洗方便，也符合大多数数据源的导出习惯。
project/ ├── data/ │ ├── raw/ │ │ ├── 000001.SZ.csv │ │ └── 600000.SH.csv │ └── processed/ │ ├── feature_2020.csv │ └── feature_2021.csv ├── src/ │ ├── data_prep.py │ ├── train_model.py │ └── backtest.py └── config.yaml如果研究的是一篮子股票的多因子模型，我更建议读入后合并成一张“长表”：每一行是“证券代码 + 日期 + 因子值 + 标签”，这样后续做横截面排序、分组回测都方便。宽表适合单标的时序预测，但多标的场景下维护成本高，稍不留神就出现索引错位。
这里有一个实操细节：数据集下载后先检查日期字段是不是datetime64，以及是否有停牌导致的缺失行。很多源码包自带的 CSV 读进来后日期是字符串，直接当索引用会排序错误。常见做法是统一转成pd.to_datetime，再按“证券代码 + 日期”排序。
2.2 标签构造：未来 N 日收益与横截面分位数
量化 ML 里的“标签”不是给行情打人工标记，而是用未来一段时间的收益算出来的。常见的有回归标签（未来 5 日收益）和分类标签（涨、跌、平）。我一般先做分类，因为分类问题对噪声的容忍度更高，模型也更容易收敛。
import pandas as pd def make_labels(df: pd.DataFrame, horizon: int = 5, method: str = "quantile", lower: float = 0.3, upper: float = 0.7): """ 为单只股票的日线数据构造未来 horizon 日收益标签。 method='quantile' 时使用时间序列分位数划分三分类； method='sign' 时直接按收益正负做二分类。 注意 df 必须已按日期升序排序。 """ df = df.copy() # 未来 horizon 日收益：第 t 日的收益是 close[t+horizon] / close[t] - 1 df["future_return"] = df["close"].shift(-horizon) / df["close"] - 1 if method == "quantile": low = df["future_return"].quantile(lower) high = df["future_return"].quantile(upper) df["label"] = 0 df.loc[df["future_return"] > high, "label"] = 1 df.loc[df["future_return"] < low, "label"] = -1 elif method == "sign": df["label"] = (df["future_return"] > 0).astype(int) else: raise ValueError("method 仅支持 quantile 或 sign") # 最后 horizon 行未来收益是 NaN，label 无效，丢掉 df = df.dropna(subset=["future_return"]) return df[["date", "close", "future_return", "label"]]这段代码有两个关键参数：horizon和分位数阈值。horizon=5表示用未来一周的收益做标签，适合中低频研究；horizon=20则接近月度。阈值上，lower=0.3, upper=0.7意味着把收益最低的 30% 标记为跌、最高的 30% 标记为涨，中间 40% 是“不动”。这样三分类样本相对平衡，比简单按 0 切分更抗噪声。
提示：上面的分位数是在整段历史上计算的，严格研究会引入轻微未来信息。更稳的做法是只在训练集区间上计算分位阈值，再把这个阈值应用到测试集。源码包如果直接对全量数据算分位数，你在复现时要留意这一层。
2.3 时间序列切分：为什么 TimeSeriesSplit 要留 gap
金融数据最忌讳随机打乱。直接用train_test_split会把未来的样本混进训练集，模型等于偷看了答案。正确做法是用TimeSeriesSplit，并且在前一段训练集和后一段测试集之间留出gap。
import pandas as pd from sklearn.model_selection import TimeSeriesSplit, train_test_split # 错误示范：随机切分 # X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2) tscv = TimeSeriesSplit(n_splits=5, test_size=250, gap=10) for fold, (train_idx, test_idx) in enumerate(tscv.split(X)): X_train, X_test = X.iloc[train_idx], X.iloc[test_idx] y_train, y_test = y.iloc[train_idx], y.iloc[test_idx] print(f"fold {fold}: train {X_train.index[0]}~{X_train.index[-1]}, " f"test {X_test.index[0]}~{X_test.index[-1]}")为什么要gap？因为如果训练集最后一天是t，测试集第一天是t+1，而标签用了未来 5 日收益，训练集最后几天的标签实际上和测试集前几天的行情重叠。gap=10表示把训练集末尾 10 个交易日去掉，从物理上切断了这种标签重叠。很多源码包只写了n_splits，不写gap，复现出来的结果虚高，这个参数是新手最容易漏掉的血泪经验。
3. 集成多种算法：线性模型、树模型和神经网络如何进同一个研究框架
标题里的“集成多种算法”不是指把所有模型预测结果一平均就完事，而是先让每个算法独立成为基线，再用投票或堆叠合成最终信号。这章给出一个能直接跑的最小模型池和集成方式。
3.1 模型池先定四个基线：线性、随机森林、XGBoost、MLP
我的选型顺序是：逻辑回归 → 随机森林 → XGBoost → MLP。逻辑回归负责测线性信号，随机森林和 XGBoost 负责抓非线性特征组合，MLP 用来验证特征之间更复杂的交互。如果 MLP 和树模型效果差不多，我倾向用树模型，因为可解释性和调参效率更高。
from sklearn.pipeline import Pipeline from sklearn.preprocessing import StandardScaler from sklearn.linear_model import LogisticRegression from sklearn.ensemble import RandomForestClassifier from sklearn.neural_network import MLPClassifier from xgboost import XGBClassifier models = { "logistic": Pipeline([ ("scaler", StandardScaler()), ("clf", LogisticRegression(C=1.0, max_iter=500)) ]), "rf": RandomForestClassifier( n_estimators=300, max_depth=5, min_samples_leaf=50, n_jobs=-1 ), "xgb": XGBClassifier( n_estimators=300, learning_rate=0.03, max_depth=4, subsample=0.8, colsample_bytree=0.8, eval_metric="logloss" ), "mlp": Pipeline([ ("scaler", StandardScaler()), ("clf", MLPClassifier( hidden_layer_sizes=(64, 32), alpha=0.01, early_stopping=True, max_iter=200 )) ]) }参数说明都写死在代码里了，但有几个值得解释：min_samples_leaf=50是控制树模型过拟合最重要的旋钮，金融数据噪声大，叶子节点太少会把个别极端行情当规律；learning_rate=0.03配合n_estimators=300是 XGBoost 的保守组合，牺牲一点训练速度换取稳定性；early_stopping=True让 MLP 在验证集不再变好时自动停，避免在噪声上学过头。
3.2 特征标准化和缺失值填充：scaler 要在训练集上 fit
很多复现项目跑出违和结果，问题不是模型，而是特征处理顺序错了。对线性模型和 MLP，标准化必须在训练集上先fit，再transform训练集和测试集，不能对全量数据一起fit_transform，否则测试集的均值方差又泄漏进训练过程。
import pandas as pd from sklearn.preprocessing import StandardScaler def prepare_features(df: pd.DataFrame, feature_cols: list, fit_scaler: bool = True, scaler=None): """ 填充缺失值并标准化。 fit_scaler=True 时，用当前数据拟合 scaler； 否则使用调用方传入的 scaler 做 transform。 """ X = df[feature_cols].copy() # 数值型缺失先填中位数；更严格的做法是按截面分行业填 X = X.fillna(X.median()) if fit_scaler and scaler is None: scaler = StandardScaler() X_scaled = scaler.fit_transform(X) elif scaler is not None: X_scaled = scaler.transform(X) else: raise ValueError("需要传一个已拟合好的 scaler，或让 fit_scaler=True") X_scaled = pd.DataFrame(X_scaled, columns=feature_cols, index=df.index) return X_scaled, scaler # 使用示例：只在训练集上 fit X_train_scaled, scaler = prepare_features(X_train, feature_cols, fit_scaler=True) X_test_scaled, _ = prepare_features(X_test, feature_cols, fit_scaler=False, scaler=scaler)这段代码的逻辑是：先把缺失值用中位数填充，再做标准化。scaler单独返回，就是让你记住“测试集只能复用训练集的统计量”。树模型对标准化不敏感，但同一套特征要喂给四个模型，统一走标准化反而省心。
3.3 把多模型合成一个结果：soft 投票与 TimeSeriesSplit 堆叠
集成最简单的落地方式是VotingClassifier，把四个模型的预测概率加权平均。金融信号里，我更推荐 soft voting，也就是对概率做平均，而不是对类别做多数表决，因为概率本身就包含模型置信度。
from sklearn.ensemble import VotingClassifier, StackingClassifier from sklearn.model_selection import TimeSeriesSplit cv_split = TimeSeriesSplit(n_splits=5, gap=10) vote_clf = VotingClassifier( estimators=list(models.items()), voting="soft", weights=[1, 1, 2, 1] )权重里 XGBoost 给 2，是因为在多数因子数据集上，XGBoost 对非线性交互的捕捉比其余三个更稳。如果你在自己数据集上发现随机森林更好，就改成[1, 2, 1, 1]，这是一个值得反复试的旋钮。
更进阶一点用堆叠：
stack_clf = StackingClassifier( estimators=list(models.items()), final_estimator=LogisticRegression(), cv=cv_split, stack_method="predict_proba" )这里最关键的是cv=cv_split，也就是堆叠内部的交叉验证也必须用TimeSeriesSplit，否则 sklearn 默认的StratifiedKFold会再次把时序数据随机打乱，等于又引入一次泄漏。堆叠的好处是让逻辑回归去学习“什么时候该信随机森林、什么时候该信 MLP”，坏处是训练开销大、更容易过拟合。小数据集上我一般只用 voting，堆叠留给特征数量多、样本量很大的场景。
4. 自己写最小回测：净值曲线、换手率和过拟合要一起看
模型输出不是研究终点。在量化研究项目里，下一步是把预测概率转成仓位，再算出净值曲线和风险指标。这章给一个几百行之内能跑通的最小回测框架，同时讲清楚为什么样本内准确率不能当结果。
4.1 从预测概率到目标仓位：阈值和持仓天数
模型测试集输出的predict_proba通常是“上涨类别”的概率。仓位规则我一般用双阈值：概率高于 0.6 做多，低于 0.4 做空，中间空仓。这样能过滤掉模型没把握的样本。
import pandas as pd def generate_positions(prob_up: pd.Series, long_thr: float = 0.6, short_thr: float = 0.4, hold_days: int = 5): """ 根据上涨概率生成目标仓位。 返回的仓位序列已经做了未来 hold_days 天的信号保持， 减少不必要的调仓。 """ raw = pd.Series(0, index=prob_up.index) raw[prob_up >= long_thr] = 1 raw[prob_up <= short_thr] = -1 # 信号保持：每 5 天只取一个仓位，避免天天换手 positions = raw.resample("5D").last().reindex(raw.index).ffill() return positions.fillna(0)hold_days这个参数很容易被忽略。很多源码包回测净值很漂亮，但拉出交易记录一看，每天调仓，换手率高得离谱。信号保持的思路是：每个持仓周期只更新一次仓位，其他时间沿用旧仓位，这是降低换手率最直接的办法。
4.2 迷你回测函数：净值、夏普、最大回撤与换手率
有了仓位序列后，用个股的日收益率和仓位做逐日收益计算。注意要用positions.shift(1)，因为今天收盘产生的信号最早只能明天开盘执行。
import numpy as np def run_backtest(returns: pd.Series, positions: pd.Series, risk_free: float = 0.02): """ returns: 标的日收益率，index 为日期 positions: 目标仓位，index 与 returns 对齐 返回净值序列和一个包含核心指标的字典。 """ # 信号滞后一日，避免用当日信号赚当日行情 strat_ret = positions.shift(1) * returns strat_ret = strat_ret.dropna() nav = (1 + strat_ret).cumprod() n_days = len(strat_ret) # 年化收益 annual_return = nav.iloc[-1] ** (252 / n_days) - 1 # 年化波动与夏普，这里无风险利率按 2% 折算 annual_vol = strat_ret.std() * np.sqrt(252) sharpe = (strat_ret.mean() * 252 - risk_free) / annual_vol # 最大回撤 drawdown = nav / nav.cummax() - 1 max_drawdown = drawdown.min() # 换手率：平均每个交易日仓位变化比例 turnover = positions.diff().abs().sum() / n_days metrics = { "annual_return": annual_return, "annual_vol": annual_vol, "sharpe": sharpe, "max_drawdown": max_drawdown, "turnover": turnover } return nav, metrics这里252是 A 股一年的交易日数量。positions.shift(1)是回测里最容易搞错的一步，少了它，回测会偷偷用到当天收盘才知道的信号，结果虚高。换手率指标不是用来好看的，它是后面判断“模型是否在疯狂追逐噪声”的关键信号。
4.3 别只用准确率说话：IC 与样本外收益
分类准确率在涨跌样本不平衡时非常骗人。比如 70% 的样本是“平”，模型全部预测“平”就有 70% 准确率，但策略一分钱赚不到。量化研究里我更习惯看 IC，也就是预测值和未来收益的秩相关系数。
from scipy.stats import spearmanr def calc_ic(positions: pd.Series, future_return: pd.Series): """ IC 表示仓位方向和未来收益的相关性。 ｜IC｜越接近 1，预测越有效；接近 0 说明信号没有信息量。 """ # 对齐日期并丢弃空值 df = pd.concat([positions, future_return], axis=1, keys=["pos", "ret"]) df = df.dropna() if len(df) < 30: return np.nan ic, _ = spearmanr(df["pos"], df["ret"]) return icIC 的阈值没有绝对标准，但在日频数据上，IC 能稳定在 0.03 以上已经算可用信号。稳定比绝对值更重要——你宁可要一个连续 12 个月 IC 都在 0.02~0.04 的策略，也不要一个月 IC 冲到 0.15、下个月直接变负的模型。黑匣子式的超参数搜索最容易在 IC 上翻车，后面避坑章节会展开。
5. 避坑 / 常见问题 / 排查：量化 ML 项目最容易翻车的 5 个地方
这章列的是我复现和自查同类源码时踩过的真实问题，每一条都按“现象 → 原因 → 解决”来写。你在跑标题这类项目时，如果结果好得不像话，先回来对照这五条。
5.1 特征里混进未来数据，回测还很漂亮
现象：训练和测试准确率都接近 80%，回测年化收益高得离谱，但一上实盘就失效。原因：某个特征计算时引用了未来数据，最常见的是对全量数据做均值、方差、分位数统计，或者用“当天收盘价”计算“当天收益”后又把它当特征。解决：把所有特征计算包进一个按日期滚动的窗口里，每一行特征只能依赖这一行及之前的数据。可以用df.expanding().mean()这类操作替代全局统计。
5.2 随机切分训练集测试集，模型在“偷看答案”
现象：用train_test_split后测试集表现很好，但按时间顺序排的样本外测试一塌糊涂。原因：随机切分把未来行情混进训练集，模型“见过”未来的波动。解决：一律用TimeSeriesSplit，并且带上gap。如果项目源码里用的是随机切分，直接把那段代码换掉，很多虚高指标会立刻现出原形。
5.3 忽略手续费和滑点，换手率把利润吃光
现象：策略净值曲线在纸面上年化 30%，但持仓周期只有 1~2 天，换手率超过 100%。原因：回测没有扣除交易成本，模型在高频追逐噪声。解决：在run_backtest中至少每次换仓扣除双边万分之三到千分之一的手续费，期货还要加上滑点。先跑出换手率，如果单边换手率超过 50%，先想办法降频，再谈收益。
5.4 全市场标签偏斜，模型只会预测“不动”
现象：模型预测结果里 90% 是 0，空仓信号极少。原因：标签构造时用了固定阈值，而股票市场大部分时间横盘，正负样本比例天然失衡。解决：改用横截面分位数标签，每个交易日只把上涨幅度最大的前 30% 标记为 1，下跌最多的前 30% 标记为 -1。如果源码数据集覆盖的股票数量够多，优先在横截面上做标签。
5.5 反复用同一段测试集调参，最后一次翻车
现象：回测指标在测试集上连续三轮提升，部署后立刻失效。原因：你已经在用测试集做决策了，测试集不再是测试集，而是第二训练集，这就是常说的“隐式过拟合”。解决：把数据切成三段：训练、验证、最终测试。日常调参只用验证集；一切定稿后，最终测试集只准跑一次。研究项目里这个纪律比任何算法都重要。
6. 最后一公里：滚动训练、特征稳定性检查与可复现实验记录
研究项目收尾前，我还会补两道工序：一是把单次训练改成滚动训练，二是给每次实验留下完整记录。很多源码包只在固定时间段上跑一次，这种做法在量化里不够可靠，因为市场风格会漂移，模型半年后可能完全失效。
6.1 滚动外推：每周重训还是每月重训
常见做法是每 20 个交易日重新训练一次模型，每次都只把最近 250 个交易日当训练集。超参数不要每次重调，固定成上一版最优值，只重训模型权重。
def rolling_train(X, y, model, retrain_interval=20, train_len=250): """在时间轴上滚动训练模型，返回每次预测概率。""" preds = [] dates = X.index start = train_len while start + retrain_interval <= len(X): train_idx = slice(start - train_len, start) test_idx = slice(start, start + retrain_interval) model_clone = model # 实际项目中用 deepcopy 或重新实例化 model_clone.fit(X.iloc[train_idx], y.iloc[train_idx]) prob = model_clone.predict_proba(X.iloc[test_idx])[:, 1] preds.append(pd.Series(prob, index=dates[test_idx])) start += retrain_interval return pd.concat(preds)这个函数的两个参数决定了策略的更新节奏：retrain_interval=20是每月调一次仓模型，train_len=250是约一年的训练窗口。滚动训练的目的不只是提升收益，而是观察模型在不同市场风格下的稳定性。
6.2 可复现实验记录：每次跑完留下哪些数字
我建议每次实验都记录成一张表，字段至少包括：模型组合、特征文件版本、训练区间、测试区间、超参数、IC、年化、最大回撤、换手率。这样即使三个月后模型失效，你也能知道是行情变了，还是某个特征处理被改坏了。
我的个人习惯是：先跑一个最简单的逻辑回归基线，再跑随机森林和 XGBoost，最后才上堆叠。如果堆叠模型不能比最好的单模型高出 20% 以上的 IC，就不要用它，复杂度本身不是优点。这个习惯帮我避开了很多“看起来高大上、实际没有边际收益”的模型堆砌。
希望这些来自真实回测坑里的经验，能帮你把这个方向的源码和数据集真正变成自己的研究体系。希望帮到你。
本文还有配套的精品资源，点击获取
