简介Adv_Fin_ML_Exercises 是一套面向量化金融与机器学习实践者的实验性习题解答取材于 Marcos Lopez De Prado《金融机器学习进展》中的选定练习适合希望把书中方法落到代码的中级学习者。内容围绕监督分类中的标注与元标注、金融交叉验证、样本权重、分数阶差分以及 Tick、成交量与美元成交额 bars 等主题展开Notebook 与配套脚本可用于复现和扩展实验。压缩包共 43 个文件约 2.8MB以 18 个 Python 脚本和 6 个 Jupyter Notebook 为核心辅以 rst 文档、Makefile、配置文件及示例数据目录涵盖 notebooks、src、data、docs、reports 等模块便于按章节与功能检索。已有 342 人学习src 中沉淀了 bars.py、snippets.py 等可复用代码既能帮助读者对照书中概念做实验也便于改造成自己的研究脚本并补充了其他 AFML 项目与文章链接供延伸阅读。1. 回测曲线漂亮却实盘亏钱这套 Jupyter Notebook 练习专攻的就是这个落差有个做量化的朋友给我看过一份回测报告年化 35%夏普 2.1最大回撤 8%。策略上线四个月净值一路往下。代码没动数据没换他想不通差在哪。我把报告拆开看问题出在三处标签按固定持有期随便定样本时间窗大幅重叠却当成独立样本喂给模型交叉验证用普通 KFold训练集里混进了验证期之后的信息。这三处正是这套围绕金融机器学习进阶教材逐个练习写解答的开源仓库反复强调的内容。仓库用 Jupyter Notebook 组织一个练习一个单元配备数据构造、标注、样本加权、交叉验证、回测评估的完整链路。适合两类人读完理论书但卡在公式到代码那一步的以及已经能写策略、但想搞清楚自己回测为什么不可信的。运行需要 Python 和数据科学基础只想跑通几段代码也足够。2. 仓库结构与技术栈Notebook 按什么逻辑切分环境怎么搭2.1 目录骨架每个练习对应一个可独立执行的 Notebook这套仓库的组织方式很朴素没有复杂包结构就是按教材的知识块切目录一个目录下面放若干.ipynb。我实际翻过之后感觉像“能单独跑的读书笔记”每个 notebook 都能从第一个单元执行到最后一个中间不依赖别的 notebook 的变量。大致结构如下fin-ml-exercises/ ├── data/ # 原始行情与中间产物全部本地缓存 │ ├── raw/ # 从数据源导出的一次性快照 │ └── processed/ # bar 构造、标签、权重落盘的位置 ├── notebooks/ │ ├── 01_bars/ # 时间棒、tick 棒、成交量棒的构造对比 │ ├── 02_labeling/ # 三重障碍、元标签、固定持有期标注 │ ├── 03_sample_weights/ # 标签重叠度、平均唯一性、时间衰减 │ ├── 04_fractional_diff/ # 分数差分与平稳性检验 │ ├── 05_cross_validation/ # Purged K-Fold、组合净化交叉验证 │ ├── 06_backtest/ # 回测统计、过拟合概率评估 │ └── 07_bet_sizing/ # 下注规模与仓位映射 ├── src/ # notebook 里反复出现的函数抽出来的公共模块 │ ├── labeling.py │ ├── sampling.py │ └── validation.py ├── requirements.txt └── README.mdnotebooks/下的编号前缀在导入时很关键。因为单元执行顺序经常被修改我通常会在开头加一段初始化代码显式把src/加入sys.path防止出现“本地能导入、别的电脑上找不到模块”的问题。2.2 环境搭建依赖不多但版本兼容性要确认清楚这套东西的依赖相对克制没有深度学习框架。核心就是数值计算和数据处理的组合。我一般按下面这个清单准备# 建议 Python 3.9 以上numba 对新版本 Python 的支持有滞后 python -m venv .venv source .venv/bin/activate # Windows 用 .venv\Scripts\activate pip install numpy pandas scipy scikit-learn matplotlib statsmodels pip install numba joblib jupyterlab几个依赖的用途分开说清楚方便你按需删减依赖在这套练习里的角色缺了会怎样numpy / pandas序列运算、时间索引对齐几乎全部单元无法运行scipy统计检验、分布拟合平稳性检验、统计量计算报错scikit-learn分类器、KFold 基类交叉验证章节直接失效statsmodels回归、ADF 检验分数差分那几节跑不完整numba / joblib多进程标注、权重计算加速大样本下耗时从分钟级涨到小时级matplotlib标签与曲线可视化图出不来不影响逻辑提示numba和 Python 小版本绑定很紧如果装完导入时报编译错误先把numba降到与当前解释器匹配的版本不要急着改业务代码。2.3 数据从哪来bar 构造是整条链路的第一道关键工序所有下游模块都依赖同一个数据结构带t1列的事件表。而事件表的起点是 bar。教材里强调过一点时间棒固定时间间隔切出来的 K 线在流动性不均的时段会丢失信息。这类练习通常会让你横向比较三种 bar时间棒每 N 分钟切一根最直观也最容易在低流动性时段出现大量噪声。tick 棒每 N 笔成交切一根把“信息到达”换成成交笔数来度量。成交量棒每 N 单位成交量切一根让每根 bar 承载差不多的换手。实践里的做法是先在notebooks/01_bars里把原始逐笔或分钟数据聚合成不同 bar再把结果存到data/processed/后续标签构造只读这份缓存。这样做的好处是标签逻辑和数据清洗彻底解耦某天想换 bar 类型只要重新跑一次 01 的 notebook02 到 06 一个字都不用改。需要注意bar 构造里最容易出问题的是时间戳语义。用左闭右开还是左开右闭会直接导致同一笔成交被算进相邻两根 bar进而让后面所有标签的整体统计产生偏差。我习惯在数据处理阶段把每根 bar 的起始时间和结束时间都显式保留宁可多两列也不要在后面靠推断。3. 核心代码链路分数差分、三重障碍标注与样本权重3.1 分数差分既想要平稳性又不想丢掉记忆金融序列做监督学习有一个长期矛盾价格本身几乎不平稳直接拿去训练模型学到的是趋势而不是规律做一阶差分变成收益率平稳了但长记忆被削掉很多有预测力的信息也没了。教材给出的折中方案是分数阶差分找一个最小的d让序列通过平稳性检验的同时保留尽可能多的原始相关性。标准的固定宽度分数差分实现核心是权重递推加法import numpy as np import pandas as pd def frac_diff_ffd(series, d, thres1e-5): series: DataFrame列为价格序列 d: 差分阶数0 d 1 是常用区间 thres: 权重截断阈值越大计算越快、精度越低 # 1) 递推生成二项式展开的权重直到绝对值小于阈值 w, k [1.0], 1 while abs(w[-1]) thres: w.append(-w[-1] / k * (d - k 1)) k 1 w np.array(w) width len(w) - 1 # 需要被消化的历史长度 out {} for col in series.columns: s series[col].ffill().dropna() if len(s) width: continue # 2) 权重反转后做 valid 卷积等价于逐窗加权求和 conv np.convolve(s.values, w[::-1], valid) out[col] pd.Series(conv, indexs.index[width:]) return pd.DataFrame(out), width这段代码有两个要点。第一权重不是预先固定长度而是靠thres动态截断——d越接近 1需要的宽度越大thres设得太小会让卷积窗口膨胀到几百个点白白消耗算力。第二np.convolve的valid模式会丢掉前width个点这几百个点没有足够的历史可供差分属于结构性缺失不能补零。参数选择上我一般这么调从d 0.1开始每次加 0.1对结果跑 ADF 检验记录第一个通过 5% 显著性水平的d再比较这个d下的序列与原始序列的相关系数。教材里提到的“最小d”就是这个交叉点——平稳性达标相关性还没掉太多。如果两者始终无法兼顾说明这段样本本身就不适合做分数差分换数据比继续调参更划算。3.2 三重障碍标注把一条价格路径转换成一个分类标签固定持有期标注的问题在于它假设“第 N 天后涨了就是正样本”完全不考虑这 N 天里发生了什么。价格先跌 8% 再拉回和一路平稳上涨标签一样风险完全不一样。三重障碍法解决的就是这个给每个样本同时设置上障碍止盈、下障碍止损和时间障碍到期看哪条先被触发。实现时通常拆成两步。第一步按波动率算动态阈值第二步扫描价格路径def daily_vol(close, span100): # 用 EWM 标准差估计日波动span 对应半衰期式的衰减 log_ret np.log(close).diff() return log_ret.ewm(spanspan).std() def triple_barrier(close, events, pt_sl(1.0, 1.0), moleculeNone): events: 索引为事件发生时间须含 t1时间障碍到期点和 trgt目标波动 pt_sl: (止盈倍数, 止损倍数)作用于 trgt 得到绝对阈值 ev events.loc[molecule] out ev[[t1]].copy(deepTrue) if pt_sl[0] 0: out[pt] pt_sl[0] * ev[trgt] # 上障碍价格偏离 else: out[pt] np.nan if pt_sl[1] 0: out[sl] -pt_sl[1] * ev[trgt] # 下障碍取负方便比大小 else: out[sl] np.nan for t0, t1 in ev[t1].items(): path close.loc[t0:t1] # 事件窗口内的价格切片 if path.empty: out.loc[t0, t1] pd.NaT continue # 相对首个价格的收益率路径 ret (path / close.loc[t0]) - 1.0 hit_pt ret[ret out.loc[t0, pt]] hit_sl ret[ret out.loc[t0, sl]] first_pt hit_pt.index[0] if len(hit_pt) else pd.NaT first_sl hit_sl.index[0] if len(hit_sl) else pd.NaT # 三条障碍比谁先到时间障碍兜底 candidates [x for x in (first_pt, first_sl, t1) if pd.notna(x)] out.loc[t0, t1] min(candidates) # 用最终触达点的收益方向打标签 out[ret] close.reindex(out[t1]).values / close.reindex(out.index).values - 1 out[bin] np.where(out[ret] 0, 1, 0) return out参数里最影响结果的是trgt。它来自daily_vol也就是对每个事件时点用当时的波动率做阈值而不是全样本一个固定百分比。这是这套方法区别于传统固定止盈止损的关键——波动大的时段阈值自动放宽波动小的时段自动收紧标签在时间维度上才有可比性。pt_sl的取值决定正负样本比例。都设 1.0 时上下对称标签接近五五开把止损设成 2.0、止盈设成 1.0等于“亏了扛、赚了跑”负样本会明显变多。这一步没有标准答案得看你的策略想在什么风险收益结构下工作。我通常先跑一次对称版看分布再根据实际交易逻辑调。3.3 样本权重重叠标签不能当独立样本计数三重障碍产出的每个样本都有一个t1也就是事件结束时间。问题在于相邻样本的[t0, t1]区间大量重叠——同一段价格走势被十几个样本反复使用模型会误以为这类行情出现了十几次。教材给出的修正叫“平均唯一性”对每个样本在它的存活区间内统计同时存活的样本数取倒数再做归一化。def num_concurrent_events(close_idx, t1, molecule): # 每个时间点上有多少个样本的事件窗口还活着 count pd.Series(0, indexclose_idx, dtypeint) for t0, t_out in t1.loc[molecule].items(): count.loc[t0:t_out] 1 return count def avg_uniqueness(t1, num_co_events, molecule): w pd.Series(indexmolecule, dtypefloat64) for t0, t_out in t1.loc[molecule].items(): # 窗口内并发数的倒数的均值就是这个样本的“独特贡献” w[t0] (1.0 / num_co_events.loc[t0:t_out]).mean() return w / w.sum() # 归一化保证权重总和为 1两个函数配合使用num_concurrent_events是全局一次的 O(N×窗口) 扫描avg_uniqueness在每个样本内部再取一次均值。计算量不小样本上万时建议用joblib按molecule分块的时间片段并行。权重算完之后把它传给分类器的sample_weight参数。这一步不做前面所有标注和验证的工作都要打个折扣因为模型实际上还是在按重叠次数加权学习。我在自己的项目里踩过一次这个坑标签逻辑改对了交叉验证也换成了净化版但权重一直没接进去回测夏普偏高 0.4 左右排查了半天才发现是这里漏了一行。4. 交叉验证与回测评估Purged K-Fold、组合净化和过拟合概率4.1 普通 KFold 为什么在金融序列上必然漏信息标准 KFold 在划分时随机打散样本这在图像、文本这类独立同分布数据上没问题。金融标签有两个特性让这个假设彻底失效一是重叠前面已经说过二是序列相关训练集里的样本和验证集里的样本在时间上挨得很近标签几乎可以互相推出来。教材给的修正叫净化purging训练集中凡是t1落在验证集时间窗内的样本全部剔除。再加一层 embargo在验证集后面再砍掉一小段时间的样本防止特征窗口跨期泄漏。class PurgedKFold: def __init__(self, n_splits5, t1None, pct_embargo0.01): self.n_splits n_splits self.t1 t1 # 每个样本的结束时间 self.pct_embargo pct_embargo # 验证集后的禁入比例 def split(self, X, yNone, groupsNone): idx X.index n len(idx) embargo int(n * self.pct_embargo) # 按时间顺序切块而不是随机打散 bounds np.linspace(0, n, self.n_splits 1).astype(int) for i in range(self.n_splits): test_start, test_end bounds[i], bounds[i 1] test_idx idx[test_start:test_end] # 1) 训练集排除与验证集时间重叠的样本 train_mask np.ones(n, dtypebool) test_t0 test_idx[0] test_t1 test_idx[-1] overlap self.t1.loc[test_t0:test_t1] train_mask[np.isin(idx, overlap.index)] False # 2) 验证集结束后追加 embargo 区一并排除 end_pos test_end embargo if end_pos n: train_mask[test_end:end_pos] False train_mask[test_start:test_end] False yield np.where(train_mask)[0], np.arange(test_start, test_end)这里最关键的是t1的传入。如果样本是三重障碍标注出来的t1就是实际触达点不是设定的时间障碍点如果用的是固定持有期t1就是t0 N。前者更准确后者只是近似。我见过有人直接拿index 5 days当t1在波动大的标的上一半以上样本的实际结束时间被算错净化等于没做。pct_embargo一般取 0.01 到 0.02。太大训练集剩不下多少太小挡不住特征泄漏。判断标准是看净化前后训练集样本数的下降幅度掉 30% 以上就得重新审视t1是不是估得太宽了。4.2 组合净化交叉验证样本不够时如何多榨出几条回测路径净化 K 折解决了泄漏但带来新问题金融数据本来就短砍掉重叠样本和 embargo 之后训练集更少回测路径只有 K 条统计上很单薄。组合净化交叉验证的思路是换一种组合方式把数据切成 N 个连续块从中选 k 个块做验证其余做训练穷举所有组合。这样能得到 C(N,k) 种切分每条切分都做净化最终能拼出多条完整回测路径。实现上不复杂核心是给每个块维护“训练集时间范围”和“验证集时间范围”然后把路径拼接起来from itertools import combinations def comb_purged_splits(index, n_blocks6, k_test2, t1None): 返回 [(train_idx, test_idx), ...]每种组合一套 n_blocks: 连续块数量 k_test: 每次拿多少块做验证 bounds np.linspace(0, len(index), n_blocks 1).astype(int) blocks [index[bounds[i]:bounds[i 1]] for i in range(n_blocks)] splits [] for test_ids in combinations(range(n_blocks), k_test): test_idx pd.Index([]).append([blocks[i] for i in test_ids]) train_idx pd.Index([]) for i in range(n_blocks): if i in test_ids: continue blk blocks[i] # 同样做净化剔除与验证集时间重叠的训练样本 if t1 is not None: kept t1.loc[blk] kept kept[(kept test_idx[0]) | (kept.index test_idx[-1])] blk kept.index train_idx train_idx.append(blk) splits.append((train_idx, test_idx)) return splitsn_blocks和k_test的组合直接决定能产生几条路径。块数越多路径越多但每块越小训练集越碎。我常用的配置是 6 块取 2 块产生 15 种划分拼出大约 5 条完整路径——足够看分布又不至于让单次训练的数据少到无法收敛。4.3 回测过拟合概率把“调参调出来的夏普”量化成一统计量这一节的方法在整套练习里最有实用价值。它回答的问题是你试了 50 组参数挑出夏普最高的那组这个夏普有多少是运气教材给出的框架是把每次试验的夏普当作来自某个分布的样本检验“真实夏普大于零”这个假设同时用有效试验次数做校正相关性高的试验不算独立计数。流程分三步记录所有试验的收益序列和夏普包括那些“失败的”配置不能只留最优的。把夏普序列标准化算出偏度和峰度用于构造经验分布。用有效试验次数计算过拟合概率值越大说明最优结果越可能是抽样噪声。from scipy.stats import norm def prob_backtest_overfit(sr_series, n_effNone): sr_series: 所有试验的夏普比率必须包含全部试验 n_eff: 有效试验次数默认用样本量近似 sr np.asarray(sr_series, dtypefloat) n len(sr) if n_eff is None else n_eff # 用偏度峰度修正标准正态逼近真实分布 m2 sr.var(ddof0) skew ((sr - sr.mean()) ** 3).mean() / m2 ** 1.5 kurt ((sr - sr.mean()) ** 4).mean() / m2 ** 2 - 3 # 期望夏普分布的标准差估计 sigma np.sqrt((1 - skew * sr.mean() (kurt / 4.0) * sr.mean() ** 2) / (n - 1)) # 最优夏普在分布中的位置越高说明越偶然 z sr.max() / sigma return norm.cdf(z) # 输出 0~1越大越可疑实际用的时候sr_series的收集方式比公式本身更重要。我的习惯是在调参脚本里把每一组参数的结果都追加到一个 CSV包括明显跑输的配置事后从这个文件里读。只记录最优结果的话这个检验没法做。返回值不到 0.5 说明结果还站得住超过 0.7 就应该怀疑——不是策略不行是你试的次数太多最优那个已经接近随机抽签。这时候该做的是减少搜索空间或者改用前面提到的组合净化路径取分布下界而不是继续调参。5. 避坑与排查Notebook 跑不通的五个高频现场5.1 环境与执行顺序类现象换了台机器ImportError: cannot import name xxx from src但 notebook 文件本身没改。原因Jupyter 的工作目录默认是 notebook 自己所在的文件夹而不是仓库根目录src不在sys.path里。另外有些人靠%cd手动切目录重启内核后又失效了。 解决在第一个单元固定加初始化代码用绝对路径推导不要靠当前工作目录的巧合。import sys, pathlib ROOT pathlib.Path.cwd().parents[1] # notebooks/xx → 仓库根 if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from src.labeling import triple_barrier现象单独跑最后一个单元没问题从头执行到中间某个单元就报KeyError。原因notebook 的单元执行顺序和文件里的顺序不一致某个变量是上次运行残留的。这在标注类 notebook 里特别常见因为中间会临时打印、注释掉几行。 解决改完逻辑后强制执行Restart Kernel and Run All。我一般同时在 notebook 顶部写一行注释标明“此文件必须顺序执行”避免几个月后回来自己也忘了。5.2 数据与逻辑类现象跑三重障碍标注结束时间t1大量为NaT或者时间障碍到期点明显早于窗口末端。原因events索引和close索引的时间粒度或者时区不一致。一边是 UTC一边是交易本地时间loc[t0:t1]切片直接空掉。 解决在数据加载入口统一做时区标准化并在构造events之前断言两边索引可切片。assert close.index.tz is None and events.index.tz is None, 时区必须统一 assert close.index.is_monotonic_increasing, 价格索引必须升序现象样本权重加起来不是 1或者分类器报sample_weight长度不匹配。原因avg_uniqueness里用的是molecule的子集索引但分类器拿到的X是全量索引顺序也不同。 解决不要直接把权重 Series 传给模型先用w.loc[X.index].values重排并且断言长度一致。这一步我踩过两次第一次报错还算好第二次没报错结果权重错位回测结果看着正常但完全是错的。现象净化交叉验证的每一折训练集都只剩很少样本模型明显欠拟合。原因t1估计得太宽或者pct_embargo设得太大两个因素叠加后重叠样本把训练集砍掉太多了。 解决先打印每折的训练集占比正常应在 60% 以上。低于这个值先把pct_embargo降到 0.005 试试再回头看t1是不是拿固定持有期算出来的——换成实际触及点通常会收窄不少。注意三重障碍的t1一旦用错下游的权重、净化、回测会同时出错但表面上看只是“效果差一点”不会抛异常。这是整套流程里最容易在无声无息中出错的地方。6. 二次开发把 Notebook 里的分析逻辑整理成可复用模块Notebook 适合探索不适合长期维护。这套练习跑完之后真正要落到自己项目得把稳定下来的函数从单元里拿出来变成模块。我做过几次这种迁移后来固定成三步走。第一步是识别“真正的函数”。判断标准很简单同一个逻辑在两个以上 notebook 里出现过或者在同一个 notebook 里被调用超过两次就值得抽出来。典型的有分数差分的权重生成、三重障碍标注、并发事件计数这几段。第二步是定接口。抽模块时最容易犯的错是把 notebook 里的全局变量一起带过去导致函数依赖外部状态。我一般要求每个函数只接受显式参数返回 DataFrame 或 Series不读全局配置。Notebook 里的形态抽成模块后的接口注意点单元格里直接改dffunc(df, ...) - DataFrame不改入参返回副本用全局close变量显式传入close便于测试和替换数据打印中间结果返回中间产物由调用方决定是否打印避免日志污染硬编码阈值作为带默认值的参数保留原来的默认值第三步是加一层配置。把波动率窗口、pt_sl、pct_embargo、块数这些参数集中到一个config.py或 YAML 里。不是为了灵活是为了让每次实验结果可复现——半年前跑出来的夏普得能靠一份配置还原出来。# config.py from dataclasses import dataclass dataclass(frozenTrue) class LabelConfig: vol_span: int 100 pt_sl: tuple (1.0, 1.0) min_ret: float 0.0 dataclass(frozenTrue) class CVConfig: n_splits: int 5 pct_embargo: float 0.01用frozenTrue是有意的防止跑到一半被某段代码改了参数实验记录和实际执行对不上。这个习惯我是在一次复盘里被逼出来的那次回测结果和记录差了很多查到最后发现是中间某个单元随手改了 embargo 没改回去白折腾了两天。最后给一个验证方法抽完模块之后用同一份数据、同一份配置分别跑 notebook 和模块版比较输出的标签分布、权重总和、每折样本量这三个指标必须完全一致。不一致就说明抽取过程中有隐式依赖没被发现。从那以后我每次做这种迁移都会先固定一份“基线输出”存成 parquet改完逐字段对比确认没问题再动下一块。希望帮到你。本文还有配套的精品资源点击获取