做量价因子做到后期很多人会碰到一个共同的瓶颈动量、波动率、均线偏离、成交集中度这些传统统计量能提取的信息越来越像互相之间相关性很高再往模子里塞新因子IC提升却越来越小。原因在于这些因子本质上都在描述行情的“数值大小”比如价格涨了多少、波动放了多少而行情本身还有一种不太容易用数值表达的信息——结构。同一个窗口里价格可以走出单边趋势也可以走出来回折返的箱体两者在收益率、波动率上可能非常接近但在“形状”上完全不一样。拓扑数据分析Topological Data AnalysisTDA恰好是研究形状和结构的一种数学框架把它用在行情因子挖掘里目的是补上传统统计量缺失的“结构通道”。这篇文章不会把TDA写成一篇数学论文而是从量化实践的角度分四步讲清楚TDA的核心概念到底是什么它和传统技术分析有什么本质差别怎样用Python从K线数据里提取拓扑结构因子以及把这些因子落地到因子库时有哪些坑。读完你会得到一套可以运行的代码而不是一堆抽象名词。我的判断是TDA不是来替代传统因子挖掘工具的它的价值在于提供一类低相关的结构化特征。它适合作为现有因子体系的一个补充模块用来捕捉“行情绕了几个圈”“结构什么时候塌缩”“支撑压力形成的拓扑边界在哪里”这类信息。正确使用它的关键不是把数学公式背下来而是理解它把K线窗口变成了什么、每个输出值到底在度量什么。1. 为什么在量化里聊TDA传统因子丢失了什么先说一个真实场景。你拿到一段30根K线的行情用传统方法计算这个窗口的特征累计涨跌幅是2.1%波动率是0.8%成交量比过去20个窗口放大1.3倍RSI在61。这些数字描述的是“这段行情有多少能量、朝哪个方向、波幅多大”。但如果你把同样的30根K线顺序打乱这些统计量几乎不变行情却可能完全不是同一个形态顺序打乱后原有的趋势结构、回调结构、箱体结构全部消失了。这说明一个严重的问题传统量价因子对时间顺序不敏感或者说对“空间形状”不敏感。再进一步想。一个V型反转和一个圆弧底在某个局部窗口里可能拥有几乎相同的收益率和波动率但在未来一段时间的延续性上完全不同。V型反转的“结构”更尖锐圆弧底的“结构”更平滑。传统因子很难区分它们因为两者的数值统计太接近但TDA可以把V型和圆弧底映射成不同的拓扑特征比如“洞”出现的位置、持续的时间、消失的方式都不一样。所以TDA在量化里的定位不是又一个预测指标而是一个描述行情几何结构的特征提取器。它把每一段行情窗口变成一个点云然后计算这个点云的拓扑不变量。这些不变量对坐标的微小扰动不敏感也就是说行情稍微抖一下特征不会剧烈跳变——这一点在因子鲁棒性上非常重要。也正因为如此TDA更适合用来挖掘那些需要“整体结构”才能表达的信号而不是简单的局部涨跌信号。2. 拓扑数据分析的核心概念从点云到持续性图要理解TDA先要理解它在做什么。它的输入可以简单理解为一个点云一堆点在某个空间里的坐标集合。在量化场景里这些点通常是K线窗口内的数据点比如每一根K线被映射成“最高价、最低价、均线偏离、成交量标准化值”这样的多维坐标。有了点云之后TDA要做的事情是随着一个过滤参数从小到大变化观察这个点云的“形状”如何出生和死亡最后用一个叫做持续性图Persistence Diagram的对象把所有结构变化记录下来。这个过程里最常听到的两个术语是单纯复形和持续同调。可以将单纯复形理解为一种“把点连接起来”的规则集合。两个点距离足够近时连一条边三个点两两距离都足够近时形成一个三角形四个点两两距离足够近时形成一个四面体。过滤参数就是那个“距离阈值”。阈值很小时边很少点云呈现出很多孤立点阈值逐渐变大边越来越多点和点连通起来形成更大的结构再继续变大三角形和更高维结构填满空洞。持续同调就是记录这些结构变化过程的工具。它区分几种维度H0连通分量也就是点云里“有多少个互不相连的团块”。H1环路或洞也就是点云里“出现了多少个围起来的圈”对应行情里的回环、震荡箱体。H2空腔对应更高维度的空结构。每个“洞”都有一个出生时间birth和死亡时间death两者的差值叫做生命跨度lifespan。生命跨度越大说明这个结构在更多尺度上都存在更可能是真实的行情结构生命跨度很小说明它只是噪声。在行情因子挖掘里最常用的是H1洞的出现次数、总生命跨度、最大生命跨度。因为趋势行情几乎没有洞箱体震荡会出现大量洞V型反转则会出现一个生命跨度极短的洞——这些正是传统因子难以刻画的结构特征。为了不让这些概念悬在空中我把TDA的核心输出和因子含义用一个表格对应起来TDA概念直观含义在行情中的对应可构造的因子过滤阈值距离多少才算“连接”不同价格尺度上的结构耦合参数扫描区间H0连通分量点云团块数量多段独立行情的数量连通分量数量变化率H1洞是否存在围起来的圈箱体、震荡、来回折返洞数量、洞总生命跨度H1出生时间洞在多大尺度出现结构开始形成的价位区间出生时间与当前价的距离H1死亡时间洞在多大尺度消失结构被突破的价位区间死亡时间与当前价的距离生命跨度结构存续的尺度范围结构强度与稳定性生命跨度均值、最大值这个表格基本可以当作后续因子设计的模板。真正落到代码时我们不需要手动实现复杂的单纯复形Python里已经有比较成熟的库可以直接调用。3. TDA在行情结构因子里的典型用法在动手写代码之前先讨论几个已经被验证比较有实践价值的用法。这些用法不是拍脑袋提出的而是从TDA的数学性质推导出来的。3.1 用H1生命跨度刻画震荡强度最直接的做法在滑动窗口内对价格点云计算H1统计每一个洞的生命跨度。箱体震荡行情中点云会形成大量短生命周期的小洞趋势行情中H1洞的数量显著减少。于是可以构造出“洞生命跨度总量”因子数值高代表行情在结构层面反复纠缠数值低代表行情结构清晰、方向性强。这个因子和波动率相关但又不完全相同。波动率只度量振幅大小而H1生命跨度还包含了“结构是否闭合”的信息。两个相同波动率的窗口H1生命跨度可以差异很大这正是TDA因子的增量信息所在。3.2 用H1出生与死亡时间定位支撑压力H1洞的出生时间代表“多远的距离尺度上开始形成环”而死亡时间代表“环在多大尺度上被填满”。在量价点云中可以将这些时间戳映射回价格坐标得到结构边界。这个边界与传统布林带、唐奇安通道有明显区别布林带是统计边界TDA边界是拓扑边界它对极端值的敏感度更低更强调“结构是否真正闭合”。由此可以构造“当前价到最近拓扑边界的距离”因子。当价格运行到拓扑边界附近时结构可能发生切换此时入场或离场的信号价值往往比较高。3.3 用多维度H0/H1组合识别结构切换把H0和H1放在一起看可以识别结构切换。比如H0从少变多意味着原来连通的行情结构正在破碎多条独立行情线并行H1从多变少意味着箱体被突破结构由一个闭合圈变成一条打开的趋势路径。这种同时变化往往发生在重大行情切换的临界区域。在因子设计中可以构造一个“结构切换度”指标使用一个窗口前后两个阶段H0、H1的差异变化量。这个指标适合做事件型信号而不是连续型因子。以上三种用法都基于同一个逻辑把行情从数值序列转成拓扑对象再对拓扑对象做统计。理解了这一点代码实现就变得清晰了。4. 环境准备与前置条件下面进入工程部分。本文所有代码使用Python实现建议使用Python 3.8及以上版本版本号以你的实际环境为准不强制某个特定版本。核心依赖如下numpy点云计算与数组操作。pandasK线数据组织和因子结果输出。ripser计算持续性图是本文的核心库。persim持续性图的可视化与后处理。scikit-learn数据标准化。matplotlib绘图辅助非必须。安装命令pip install numpy pandas ripser persim scikit-learn matplotlib如果网络环境不方便安装ripser这类依赖也可以先看第5.1节的最小版实现理解原理后再回来用完整库。ripser底层是C实现计算速度比纯Python暴力实现快很多实际项目建议直接使用。本节不需要复杂的IDE配置一个能跑Jupyter Notebook的环境或者命令行Python环境都可以。代码文件结构建议按下面这样组织便于后续把因子接入策略回测系统tda_quant/ ├── data/ │ └── stock_data.csv ├── src/ │ ├── __init__.py │ └── tda_factor.py ├── scripts/ │ └── run_tda_factor.py └── output/ └── tda_factor.csv在动手写特征提取之前首要任务是把K线数据组织成“点云”。这一步直接决定因子质量值得单独花一节详细说。5. 核心流程拆解从K线到点云再到因子完整的TDA因子提取流程可以拆成五个步骤。每一步都有独立的目标和容易踩坑的地方。5.1 数据清洗与对齐首先确保K线数据按时间升序排列没有重复时间戳没有明显的脏数据。这一步不可省略因为点云计算对顺序很敏感如果数据错位计算出来的拓扑结构完全没有意义。# 文件路径scripts/run_tda_factor.py import pandas as pd df pd.read_csv(data/stock_data.csv, parse_dates[datetime]) df df.sort_values(datetime).drop_duplicates(subsetdatetime).reset_index(dropTrue) df df.dropna(subset[open, high, low, close, volume])这里drop_duplicates保证同一时间戳只保留一行dropna剔除缺失K线。实际操作中如果某根K线的最高价低于最低价也应当作为脏数据处理但真实数据源一般不太可能出现这种情况所以代码里可以只做基础校验。5.2 构造点云特征点云不是直接把K线的原始值丢进去。原始最高价、最低价、成交量的量纲完全不同最高价可能在3000左右成交量可能在上百万直接把它们作为坐标距离计算会被量纲主导拓扑结构信息被淹没。推荐的做法是把每个时间点的多维度信息编码成一个高维坐标点。这里给出一组常见的映射方式维度1当前K线收盘价相对前N根K线收盘价的移动平均偏离描述价格水平。维度2最高价与最低价的价差描述K线振幅。维度3成交量相对过去N根K线成交量的Z分数描述量能变化。维度4当前K线相对前一根K线的收益率描述短期方向。维度5收盘价在最近N根K线区间中的百分位位置描述价格在区间中的位置。这些特征计算完之后要做一个关键操作滚动标准化或全局标准化但绝对不能全量标准化。全部数据一起标准化时当前窗口的特征会用到未来数据的信息这在因子回测里是典型的未来函数会虚高因子表现。正确做法是用滚动窗口计算均值与标准差再用当前值做标准化。简单起见本文示例使用StandardScaler仅对整个数据集做演示但要明确这在实际回测中是需要避免的。得到特征矩阵后直接作为点云输入TDA计算。# 文件路径src/tda_factor.py import numpy as np import pandas as pd def build_point_cloud(df, window20): close df[close].values high df[high].values low df[low].values volume df[volume].values close_ma pd.Series(close).rolling(window).mean().values volume_mean pd.Series(volume).rolling(window).mean().values volume_std pd.Series(volume).rolling(window).std().values rolling_min pd.Series(low).rolling(window).min().values rolling_max pd.Series(high).rolling(window).max().values feat1 (close - close_ma) / np.where(close_ma 0, 1e-12, close_ma) feat2 high - low feat3 (volume - volume_mean) / np.where(volume_std 0, 1e-12, volume_std) feat4 np.append([0.0], np.diff(close) / np.where(close[:-1] 0, 1e-12, close[:-1])) feat5 (close - rolling_min) / np.where((rolling_max - rolling_min) 0, 1e-12, rolling_max - rolling_min) X np.column_stack([feat1, feat2, feat3, feat4, feat5]) return X这里有一个值得注意的细节rolling(window).min()和rolling(window).max()会用未来数据因为在时间t时滚动窗口默认包含t及之前的window-1个数据不包含未来数据所以这里没有未来函数问题。但volume_mean的使用需要注意窗口的对齐方式rolling(window).mean()在当前时间点使用的是当前时间点及过去window-1个数据所以也不包含未来数据。5.3 计算持续性图点云构造好之后调用ripser计算持续性图。ripser的输入是二维数组每一行是一个点列是坐标维度。它返回一个包含不同维度的持续同调信息的列表dgms[0]是H0的结果dgms[1]是H1的结果。# 文件路径src/tda_factor.py from ripser import ripser def compute_persistence(X, maxdim1): result ripser(X, maxdimmaxdim) return result[dgms]ripser返回的H1矩阵里每一行是一组(birth, death)。但注意并不是所有洞都会在有限阈值内死亡。有些洞在过滤结束时仍然存在对应death是inf。在计算因子时要区分处理这些“未死亡”的结构可以直接丢弃也可以把inf替换为点云中所有点对距离的最大值。具体做法在后面的完整代码里演示。5.4 从持续性图提取因子有了持续性图矩阵之后因子提取就是把矩阵压缩成一个或几个标量。这里的关键是不要只取一个指标而是设计一组覆盖不同结构角度的指标。常用指标组合如下H1洞数量持续性图中有效洞的总数。H1总生命跨度所有洞里(death - birth)之和越大说明震荡结构越强。H1最大生命跨度最显著的那个结构持续多久。H1出生时间均值洞出现的尺度位置。H1死亡时间均值洞消失的尺度位置。H1生命跨度偏度洞的生命跨度分布是否偏向长尾。这组指标可以合成为一个因子向量后续既可以单独使用也可以作为机器学习模型的输入特征。5.5 滑动窗口与特征序列因子必须在时间序列上连续生成所以要采用滑动窗口方式。对每个窗口重复步骤5.2到5.4得到该窗口的因子值然后把因子值与窗口末尾的时间戳对齐。这样每个时间点都有一个由过去N根K线计算出的拓扑因子。窗口大小和滑动步长是TDA因子中最关键的超参数。窗口太小点云数量太少拓扑结构不稳定窗口太大Rips复形的计算量显著增加而且行情结构已经被平均化失去局部意义。从实践角度看5分钟K线或1分钟K线场景窗口取30到90比较合理日线场景窗口取60到120比较合理。具体数值需要结合标的和周期做参数扫描验证。6. Rips复形的复杂度窗口为什么窗口不能无限大有一个问题很多人会忽略Rips复形的计算复杂度非常高。假设点云里有N个点Rips过滤会枚举所有点对、所有三角形甚至更高维的单形。点对数量是O(N^2)三角形数量是O(N^3)如果计算到维度d复杂度接近O(N^(d1))。这个增长非常快。因此在实际计算中窗口大小不能随意设置。一个窗口取30个点点对只有435个取100个点点对就有4950个而三角形数量会达到16万级别。当窗口扩大到200个点时点对约2万个三角形约131万个计算耗时上升非常明显。解决这个问题的常用办法有三种控制窗口大小把窗口控制在30到60个数据点这是最直接的办法。降采样窗口内原始K线数量很多时先做等间隔降采样减少点云数量。比如原来1000根1分钟K线每5根取1根得到200个点再计算TDA。维度削减点云维度不是越高越好。经验上3到6维就能保留足够结构信息再高的维度只会增加计算开销对因子增量信息贡献有限。理解这个复杂度问题对后面实际跑数据非常有帮助。如果你发现代码运行很慢第一反应不是优化代码而是检查窗口大小和点云维度。7. 完整示例TDA行情结构因子挖掘代码下面给出一个完整的、可以直接运行的示例。先给出一个纯Python暴力计算版本帮你理解TDA内部发生了什么然后给出基于ripser的生产版本用于实际因子计算。7.1 纯Python暴力版理解TDA的边界这个版本不依赖TDA库只用itertools和numpy计算三个点形成的最小外接圆半径再统计所有三角形的半径分布。虽然这个版本不能计算完整的持续性图但能让你直观理解“洞”是怎么随距离阈值出现和消失的。# 文件路径demo/tda_bruteforce.py import itertools import numpy as np def circum_radius(p, q, r): 计算三个点的外接圆半径如果三点共线则返回最大值的一半 a np.linalg.norm(p - q) b np.linalg.norm(q - r) c np.linalg.norm(r - p) s (a b c) / 2.0 area_sq max(s * (s - a) * (s - b) * (s - c), 0.0) denom 2.0 * np.sqrt(area_sq) if denom 1e-12: return max(a, b, c) / 2.0 return (a * b * c) / denom def brute_force_tda(points): 暴力枚举所有三角形返回外接圆半径分布作为结构特征 triangles [] n len(points) for comb in itertools.combinations(range(n), 3): p points[comb[0]] q points[comb[1]] r points[comb[2]] radius circum_radius(p, q, r) triangles.append(radius) triangles np.array(triangles) return { triangle_count: len(triangles), mean_circum_radius: triangles.mean() if len(triangles) else 0.0, std_circum_radius: triangles.std() if len(triangles) else 0.0, max_circum_radius: triangles.max() if len(triangles) else 0.0, } if __name__ __main__: rng np.random.default_rng(42) # 生成一组接近圆环的点云模拟箱体震荡 theta np.linspace(0, 2 * np.pi, 30) circle np.column_stack([np.cos(theta), np.sin(theta)]) circle rng.normal(0, 0.03, sizecircle.shape) feats brute_force_tda(circle) print(feats)这个版本的输出是三个点形成的外接圆半径统计量。箱体行情中外接圆半径分布集中趋势行情中点云沿一条线展开外接圆半径会比较分散。这个直觉就是TDA结构因子的雏形。7.2 基于ripser的生产版本生产版本的核心是TDAFactor类它封装了数据加载、点云构建、持续性图计算和因子提取四个流程。# 文件路径src/tda_factor.py import numpy as np import pandas as pd from ripser import ripser from sklearn.preprocessing import StandardScaler class TDAFactor: def __init__(self, window60, stride5, maxdim1, scaleTrue): self.window window self.stride stride self.maxdim maxdim self.scale scale def _build_point_cloud(self, df): close df[close].values.astype(float) high df[high].values.astype(float) low df[low].values.astype(float) volume df[volume].values.astype(float) close_ma pd.Series(close).rolling(self.window).mean().values volume_mean pd.Series(volume).rolling(self.window).mean().values volume_std pd.Series(volume).rolling(self.window).std().values rolling_max pd.Series(high).rolling(self.window).max().values rolling_min pd.Series(low).rolling(self.window).min().values feat1 (close - close_ma) / np.where(np.abs(close_ma) 1e-12, 1e-12, close_ma) feat2 high - low feat3 (volume - volume_mean) / np.where(np.abs(volume_std) 1e-12, 1e-12, volume_std) feat4 np.append([0.0], np.diff(close) / np.where(np.abs(close[:-1]) 1e-12, 1e-12, close[:-1])) feat5 (close - rolling_min) / np.where( np.abs(rolling_max - rolling_min) 1e-12, 1e-12, rolling_max - rolling_min ) X np.column_stack([feat1, feat2, feat3, feat4, feat5]) # 去掉滚动窗口未形成的前几行 X[:self.window - 1] np.nan return X def _persistence_features(self, X): result ripser(X, maxdimself.maxdim) dgms result[dgms] h0 dgms[0] if len(dgms) 0 else np.empty((0, 2)) h1 dgms[1] if len(dgms) 1 else np.empty((0, 2)) # 过滤 inf 死亡点用点云最大成对距离替代 finite_h1 h1[np.isfinite(h1[:, 1])] if len(finite_h1) 0: finite_h1 np.empty((0, 2)) if len(finite_h1) 0: lifespans_h1 finite_h1[:, 1] - finite_h1[:, 0] total_life_h1 float(lifespans_h1.sum()) max_life_h1 float(lifespans_h1.max()) mean_life_h1 float(lifespans_h1.mean()) std_life_h1 float(lifespans_h1.std()) birth_mean_h1 float(finite_h1[:, 0].mean()) death_mean_h1 float(finite_h1[:, 1].mean()) else: total_life_h1 0.0 max_life_h1 0.0 mean_life_h1 0.0 std_life_h1 0.0 birth_mean_h1 0.0 death_mean_h1 0.0 finite_h0 h0[np.isfinite(h0[:, 1])] if len(finite_h0) 0: lifespans_h0 finite_h0[:, 1] - finite_h0[:, 0] total_life_h0 float(lifespans_h0.sum()) else: total_life_h0 0.0 return { h1_count: len(finite_h1), h1_total_life: total_life_h1, h1_max_life: max_life_h1, h1_mean_life: mean_life_h1, h1_std_life: std_life_h1, h1_birth_mean: birth_mean_h1, h1_death_mean: death_mean_h1, h0_total_life: total_life_h0 } def extract(self, df): X self._build_point_cloud(df) n len(df) records [] start self.window - 1 for i in range(start, n - self.window 1, self.stride): window_X X[i - self.window 1: i 1] if np.any(np.isnan(window_X)): continue feats self._persistence_features(window_X) feats[datetime] df[datetime].iloc[i] feats[close] df[close].iloc[i] records.append(feats) feature_df pd.DataFrame(records) return feature_df这段代码里比较关键的设计有三个一是滚动窗口的特征标准化。这里的_build_point_cloud使用滚动窗口计算MA和标准差而不是全局统计量降低未来函数风险。需要说明的是为了演示简洁这里没有对特征做归一化到统一尺度实际项目建议在窗口内部再做一次标准归一化。二是inf死亡点的处理。ripser在过滤结束时可能仍有洞未死亡返回deathinf。这里把它们过滤掉只统计有限生命跨度的结构。如果某个窗口所有洞都是无穷生命跨度那就说明该窗口点云没有形成任何闭合空间这本身也是有效信息。三是滑动窗口的对齐。使用stride控制计算频率时间戳对齐到窗口最后一个K线。这样可以控制因子计算频率也方便合并到下游策略。7.3 主程序运行主程序做的事情是读取数据创建TDAFactor对象调用extract方法得到因子序列并保存到CSV文件。# 文件路径scripts/run_tda_factor.py import pandas as pd from src.tda_factor import TDAFactor df pd.read_csv(data/stock_data.csv, parse_dates[datetime]) df df.sort_values(datetime).drop_duplicates(subsetdatetime).reset_index(dropTrue) # 如果数据不是按时间升序先排序 tda TDAFactor(window60, stride5, maxdim1) feature_df tda.extract(df) print(feature_df.head()) print(feature_df.describe()) output_path output/tda_factor.csv feature_df.to_csv(output_path, indexFalse) print(ffactor saved to {output_path})运行命令cd tda_quant python scripts/run_tda_factor.py预期会看到类似下面的输出列名一致具体数值取决于数据datetime close h1_count h1_total_life ... 0 2024-01-08 10:30:00 3021.50 2 0.312 ... 1 2024-01-08 10:35:00 3025.80 3 0.458 ... ...这说明每个时间点都对应一组拓扑结构因子。接下来要做的是因子验证。8. 运行结果与效果验证怎么判断因子有没有价值构造完因子后第一件事不是直接扔进回测而是做三类验证结构合理性、相关性检验、预测能力检验。8.1 结构合理性检查先看因子的时间序列分布是否符合直觉。可以绘制因子值和行情K线在同一张图上观察箱体震荡区间H1洞数量是否偏高、趋势段H1洞数量是否偏低。这个观察性检查能快速发现代码逻辑错误。import matplotlib.pyplot as plt fig, ax1 plt.subplots(figsize(12, 6)) ax1.plot(df[datetime], df[close], colorblack, alpha0.7, labelclose) ax2 ax1.twinx() ax2.plot(pd.to_datetime(feature_df[datetime]), feature_df[h1_count], colorred, alpha0.7, labelh1_count) plt.title(TDA Factor H1 Count vs Price) plt.legend() plt.show()如果发现趋势行情里H1数量反而很高那大概率是点云构造方式有问题比如特征维度选择不当导致距离度量失真。8.2 与已有因子的相关性检验把TDA因子和已有传统因子放在一起计算相关性矩阵。理想情况下TDA因子与传统动量、波动率因子的相关系数应该在0.2以下与成交量类因子的相关系数也应当较低。如果相关系数过高说明TDA因子捕获的仍然是传统信息增量有限。8.3 预测能力检验一个常用的轻量级验证方法是计算因子值与下一期收益的秩相关系数IC。具体做法是把特征按时间顺序切片下一期收益用未来N根K线的累计收益计算Spearman秩相关。import numpy as np from scipy import stats feature_df feature_df.sort_values(datetime).reset_index(dropTrue) close_map pd.Series(df[close].values, indexpd.to_datetime(df[datetime])).sort_index() aligned_close close_map.reindex(pd.to_datetime(feature_df[datetime])).values future_return pd.Series(aligned_close).shift(-5) / aligned_close - 1.0 valid future_return.notna() ic, p_value stats.spearmanr(feature_df[h1_total_life][valid], future_return[valid]) print(fIC: {ic:.4f}, p-value: {p_value:.4f})注意这里的IC计算只是示例真正做因子研究时需要更严谨的分组检验、分层回测和样本外验证不能因为单段IC高就认定因子有效。9. 常见问题与排查方法我在写TDA因子代码时遇到过几个比较典型的坑整理成排查表供你对照问题现象可能原因排查方式解决方案ripser运行非常慢窗口过大点云数量过多统计点云规模测试不同窗口耗时降采样或减小窗口点云维度控制在6维以内返回的H1矩阵为空点云形状简单没有形成闭合洞检查点云是否接近一条直线或集中在极小区域增加点云复杂度调整特征构造大量deathinf过滤参数达到最大值时洞仍未闭合查看点云最大成对距离过滤inf或用最大成对距离替代特征值出现NaN滚动窗口早期数据不足或分母为0打印特征矩阵检查rolling对齐丢弃窗口未形成的行分母加微小量因子值与行情明显不符点云特征标准化方式错误对照K线图观察检查是否错误地使用了全样本标准化IC方向不稳定窗口参数不合适或特征构造不稳定做参数敏感性分析使用多个窗口做因子合成而非单窗口因子与传统因子高度相关点云特征被某个维度主导检查特征维度方差贡献增加维度或改用更平衡的特征组合其中最常见的还是第一个计算慢。Rips复形虽然是目前计算持续同调的高效算法但它仍然需要枚举大量单形点云规模稍微上去计算开销就会指数增长。生产环境下建议先用一个小数据子集跑通流程确认因子逻辑正确后再考虑用并行或抽样方式扩大计算规模。10. 最佳实践与工程建议这部分是实践经验的总结。如果要把TDA因子真正用到研究或生产环境下面几条建议值得认真考虑。10.1 参数不要一次定死窗口大小、滑动步长、点云维度、特征选择这些参数对因子表现影响很大。建议设计一组参数网格做敏感性分析看因子IC的均值和稳定性。如果因子IC只有在某一组精确参数下才有效在其他参数下全部失效那这个因子大概率是过拟合的。理想情况是在参数网格的一个较广范围内因子方向一致IC保持在合理水平。10.2 警惕未来函数TDA计算中最大的未来函数风险不在ripser本身而在特征构造。比如用全局的均值和标准差做标准化就会把未来数据泄露到当前特征中。正确的做法是用滚动窗口计算或者先对数据集做时间切分在训练集上拟合标准化参数再应用到测试集。回测系统里尤其要注意这一点。10.3 因子入库前要做完整校验如果把因子落库到因子库或数据库必须遵循合法授权、测试环境验证、备份、回滚和最小权限原则。具体来说先在测试库执行建表和写入脚本确认字段类型、空值比例、时间对齐都正确生产库执行前备份相关表写入因子数据时只使用最小权限账号禁用DROP、TRUNCATE等高危操作。因子数据一旦污染影响的是后续所有策略所以宁可慢一点也不能跳过验证。10.4 计算TDA特征时的资源控制TDA特征计算属于计算密集型任务不适合在主线程回测循环里直接调用。建议在因子预处理阶段一次性计算并保存然后回测时读取缓存。如果标的多、频率高可以按标的并行计算或者用采样点云近似计算不必对每一根K线都做完整TDA。10.5 多窗口合成比单窗口更稳健单窗口TDA特征容易受噪声和参数影响。一种更稳的做法是同时计算多个窗口比如30、60、120的TDA特征然后把同一类特征做标准化后取平均。这类似多尺度特征融合能让因子保留不同级别行情结构的信息降低单窗口参数敏感性。10.6 别忽视可解释性TDA因子虽然来自数学概念但并不意味着必须黑箱使用。每一步都可以加可解释的约束比如窗口内点云可视化、持续性图可视化、结构切换事件与行情事件的对照。保留这些信息后续做因子归因和策略复盘时非常有价值。11. 总结与后续方向至此你应该已经理解TDA在行情结构因子挖掘中的定位它不是替代传统的动量、波动率、成交量因子而是补充一类传统统计量捕捉不到的“结构维度”信息。核心流程可以归结为四步构造点云、计算持续性图、提取拓扑特征、做因子验证。本文给出了从数据准备到因子落地的完整Python代码你在自己的数据上跑通之后可以把H1生命跨度、洞的数量等特征作为新的特征列加入现有因子库然后与传统因子做相关性分析看看真正增加了多少信息。后续可以深入的方向包括持续性图的高阶统计量比如持续熵、持续景观的L2范数把TDA特征作为图神经网络的节点特征以及多标的多周期的TDA因子合成。这些方向都在朝着同一个目标前进让机器能够更完整地理解行情的形状而不只是行情的数字。如果这篇文章对你有帮助建议收藏备用后续做因子挖掘时可以随时对照代码结构复用。有一个善意的提醒TDA因子的有效性高度依赖标的、周期和数据质量从一个市场得出的规律不一定能迁移到另一个市场。无论因子表现多好都要先在样本外、在严格的回测框架里验证再考虑实盘使用。量化研究里的每一步都要对风险保持足够的敬畏。 拓扑数据分析TDA：从点云到持续性图，挖掘低相关行情结构因子
做量价因子做到后期很多人会碰到一个共同的瓶颈动量、波动率、均线偏离、成交集中度这些传统统计量能提取的信息越来越像互相之间相关性很高再往模子里塞新因子IC提升却越来越小。原因在于这些因子本质上都在描述行情的“数值大小”比如价格涨了多少、波动放了多少而行情本身还有一种不太容易用数值表达的信息——结构。同一个窗口里价格可以走出单边趋势也可以走出来回折返的箱体两者在收益率、波动率上可能非常接近但在“形状”上完全不一样。拓扑数据分析Topological Data AnalysisTDA恰好是研究形状和结构的一种数学框架把它用在行情因子挖掘里目的是补上传统统计量缺失的“结构通道”。这篇文章不会把TDA写成一篇数学论文而是从量化实践的角度分四步讲清楚TDA的核心概念到底是什么它和传统技术分析有什么本质差别怎样用Python从K线数据里提取拓扑结构因子以及把这些因子落地到因子库时有哪些坑。读完你会得到一套可以运行的代码而不是一堆抽象名词。我的判断是TDA不是来替代传统因子挖掘工具的它的价值在于提供一类低相关的结构化特征。它适合作为现有因子体系的一个补充模块用来捕捉“行情绕了几个圈”“结构什么时候塌缩”“支撑压力形成的拓扑边界在哪里”这类信息。正确使用它的关键不是把数学公式背下来而是理解它把K线窗口变成了什么、每个输出值到底在度量什么。1. 为什么在量化里聊TDA传统因子丢失了什么先说一个真实场景。你拿到一段30根K线的行情用传统方法计算这个窗口的特征累计涨跌幅是2.1%波动率是0.8%成交量比过去20个窗口放大1.3倍RSI在61。这些数字描述的是“这段行情有多少能量、朝哪个方向、波幅多大”。但如果你把同样的30根K线顺序打乱这些统计量几乎不变行情却可能完全不是同一个形态顺序打乱后原有的趋势结构、回调结构、箱体结构全部消失了。这说明一个严重的问题传统量价因子对时间顺序不敏感或者说对“空间形状”不敏感。再进一步想。一个V型反转和一个圆弧底在某个局部窗口里可能拥有几乎相同的收益率和波动率但在未来一段时间的延续性上完全不同。V型反转的“结构”更尖锐圆弧底的“结构”更平滑。传统因子很难区分它们因为两者的数值统计太接近但TDA可以把V型和圆弧底映射成不同的拓扑特征比如“洞”出现的位置、持续的时间、消失的方式都不一样。所以TDA在量化里的定位不是又一个预测指标而是一个描述行情几何结构的特征提取器。它把每一段行情窗口变成一个点云然后计算这个点云的拓扑不变量。这些不变量对坐标的微小扰动不敏感也就是说行情稍微抖一下特征不会剧烈跳变——这一点在因子鲁棒性上非常重要。也正因为如此TDA更适合用来挖掘那些需要“整体结构”才能表达的信号而不是简单的局部涨跌信号。2. 拓扑数据分析的核心概念从点云到持续性图要理解TDA先要理解它在做什么。它的输入可以简单理解为一个点云一堆点在某个空间里的坐标集合。在量化场景里这些点通常是K线窗口内的数据点比如每一根K线被映射成“最高价、最低价、均线偏离、成交量标准化值”这样的多维坐标。有了点云之后TDA要做的事情是随着一个过滤参数从小到大变化观察这个点云的“形状”如何出生和死亡最后用一个叫做持续性图Persistence Diagram的对象把所有结构变化记录下来。这个过程里最常听到的两个术语是单纯复形和持续同调。可以将单纯复形理解为一种“把点连接起来”的规则集合。两个点距离足够近时连一条边三个点两两距离都足够近时形成一个三角形四个点两两距离足够近时形成一个四面体。过滤参数就是那个“距离阈值”。阈值很小时边很少点云呈现出很多孤立点阈值逐渐变大边越来越多点和点连通起来形成更大的结构再继续变大三角形和更高维结构填满空洞。持续同调就是记录这些结构变化过程的工具。它区分几种维度H0连通分量也就是点云里“有多少个互不相连的团块”。H1环路或洞也就是点云里“出现了多少个围起来的圈”对应行情里的回环、震荡箱体。H2空腔对应更高维度的空结构。每个“洞”都有一个出生时间birth和死亡时间death两者的差值叫做生命跨度lifespan。生命跨度越大说明这个结构在更多尺度上都存在更可能是真实的行情结构生命跨度很小说明它只是噪声。在行情因子挖掘里最常用的是H1洞的出现次数、总生命跨度、最大生命跨度。因为趋势行情几乎没有洞箱体震荡会出现大量洞V型反转则会出现一个生命跨度极短的洞——这些正是传统因子难以刻画的结构特征。为了不让这些概念悬在空中我把TDA的核心输出和因子含义用一个表格对应起来TDA概念直观含义在行情中的对应可构造的因子过滤阈值距离多少才算“连接”不同价格尺度上的结构耦合参数扫描区间H0连通分量点云团块数量多段独立行情的数量连通分量数量变化率H1洞是否存在围起来的圈箱体、震荡、来回折返洞数量、洞总生命跨度H1出生时间洞在多大尺度出现结构开始形成的价位区间出生时间与当前价的距离H1死亡时间洞在多大尺度消失结构被突破的价位区间死亡时间与当前价的距离生命跨度结构存续的尺度范围结构强度与稳定性生命跨度均值、最大值这个表格基本可以当作后续因子设计的模板。真正落到代码时我们不需要手动实现复杂的单纯复形Python里已经有比较成熟的库可以直接调用。3. TDA在行情结构因子里的典型用法在动手写代码之前先讨论几个已经被验证比较有实践价值的用法。这些用法不是拍脑袋提出的而是从TDA的数学性质推导出来的。3.1 用H1生命跨度刻画震荡强度最直接的做法在滑动窗口内对价格点云计算H1统计每一个洞的生命跨度。箱体震荡行情中点云会形成大量短生命周期的小洞趋势行情中H1洞的数量显著减少。于是可以构造出“洞生命跨度总量”因子数值高代表行情在结构层面反复纠缠数值低代表行情结构清晰、方向性强。这个因子和波动率相关但又不完全相同。波动率只度量振幅大小而H1生命跨度还包含了“结构是否闭合”的信息。两个相同波动率的窗口H1生命跨度可以差异很大这正是TDA因子的增量信息所在。3.2 用H1出生与死亡时间定位支撑压力H1洞的出生时间代表“多远的距离尺度上开始形成环”而死亡时间代表“环在多大尺度上被填满”。在量价点云中可以将这些时间戳映射回价格坐标得到结构边界。这个边界与传统布林带、唐奇安通道有明显区别布林带是统计边界TDA边界是拓扑边界它对极端值的敏感度更低更强调“结构是否真正闭合”。由此可以构造“当前价到最近拓扑边界的距离”因子。当价格运行到拓扑边界附近时结构可能发生切换此时入场或离场的信号价值往往比较高。3.3 用多维度H0/H1组合识别结构切换把H0和H1放在一起看可以识别结构切换。比如H0从少变多意味着原来连通的行情结构正在破碎多条独立行情线并行H1从多变少意味着箱体被突破结构由一个闭合圈变成一条打开的趋势路径。这种同时变化往往发生在重大行情切换的临界区域。在因子设计中可以构造一个“结构切换度”指标使用一个窗口前后两个阶段H0、H1的差异变化量。这个指标适合做事件型信号而不是连续型因子。以上三种用法都基于同一个逻辑把行情从数值序列转成拓扑对象再对拓扑对象做统计。理解了这一点代码实现就变得清晰了。4. 环境准备与前置条件下面进入工程部分。本文所有代码使用Python实现建议使用Python 3.8及以上版本版本号以你的实际环境为准不强制某个特定版本。核心依赖如下numpy点云计算与数组操作。pandasK线数据组织和因子结果输出。ripser计算持续性图是本文的核心库。persim持续性图的可视化与后处理。scikit-learn数据标准化。matplotlib绘图辅助非必须。安装命令pip install numpy pandas ripser persim scikit-learn matplotlib如果网络环境不方便安装ripser这类依赖也可以先看第5.1节的最小版实现理解原理后再回来用完整库。ripser底层是C实现计算速度比纯Python暴力实现快很多实际项目建议直接使用。本节不需要复杂的IDE配置一个能跑Jupyter Notebook的环境或者命令行Python环境都可以。代码文件结构建议按下面这样组织便于后续把因子接入策略回测系统tda_quant/ ├── data/ │ └── stock_data.csv ├── src/ │ ├── __init__.py │ └── tda_factor.py ├── scripts/ │ └── run_tda_factor.py └── output/ └── tda_factor.csv在动手写特征提取之前首要任务是把K线数据组织成“点云”。这一步直接决定因子质量值得单独花一节详细说。5. 核心流程拆解从K线到点云再到因子完整的TDA因子提取流程可以拆成五个步骤。每一步都有独立的目标和容易踩坑的地方。5.1 数据清洗与对齐首先确保K线数据按时间升序排列没有重复时间戳没有明显的脏数据。这一步不可省略因为点云计算对顺序很敏感如果数据错位计算出来的拓扑结构完全没有意义。# 文件路径scripts/run_tda_factor.py import pandas as pd df pd.read_csv(data/stock_data.csv, parse_dates[datetime]) df df.sort_values(datetime).drop_duplicates(subsetdatetime).reset_index(dropTrue) df df.dropna(subset[open, high, low, close, volume])这里drop_duplicates保证同一时间戳只保留一行dropna剔除缺失K线。实际操作中如果某根K线的最高价低于最低价也应当作为脏数据处理但真实数据源一般不太可能出现这种情况所以代码里可以只做基础校验。5.2 构造点云特征点云不是直接把K线的原始值丢进去。原始最高价、最低价、成交量的量纲完全不同最高价可能在3000左右成交量可能在上百万直接把它们作为坐标距离计算会被量纲主导拓扑结构信息被淹没。推荐的做法是把每个时间点的多维度信息编码成一个高维坐标点。这里给出一组常见的映射方式维度1当前K线收盘价相对前N根K线收盘价的移动平均偏离描述价格水平。维度2最高价与最低价的价差描述K线振幅。维度3成交量相对过去N根K线成交量的Z分数描述量能变化。维度4当前K线相对前一根K线的收益率描述短期方向。维度5收盘价在最近N根K线区间中的百分位位置描述价格在区间中的位置。这些特征计算完之后要做一个关键操作滚动标准化或全局标准化但绝对不能全量标准化。全部数据一起标准化时当前窗口的特征会用到未来数据的信息这在因子回测里是典型的未来函数会虚高因子表现。正确做法是用滚动窗口计算均值与标准差再用当前值做标准化。简单起见本文示例使用StandardScaler仅对整个数据集做演示但要明确这在实际回测中是需要避免的。得到特征矩阵后直接作为点云输入TDA计算。# 文件路径src/tda_factor.py import numpy as np import pandas as pd def build_point_cloud(df, window20): close df[close].values high df[high].values low df[low].values volume df[volume].values close_ma pd.Series(close).rolling(window).mean().values volume_mean pd.Series(volume).rolling(window).mean().values volume_std pd.Series(volume).rolling(window).std().values rolling_min pd.Series(low).rolling(window).min().values rolling_max pd.Series(high).rolling(window).max().values feat1 (close - close_ma) / np.where(close_ma 0, 1e-12, close_ma) feat2 high - low feat3 (volume - volume_mean) / np.where(volume_std 0, 1e-12, volume_std) feat4 np.append([0.0], np.diff(close) / np.where(close[:-1] 0, 1e-12, close[:-1])) feat5 (close - rolling_min) / np.where((rolling_max - rolling_min) 0, 1e-12, rolling_max - rolling_min) X np.column_stack([feat1, feat2, feat3, feat4, feat5]) return X这里有一个值得注意的细节rolling(window).min()和rolling(window).max()会用未来数据因为在时间t时滚动窗口默认包含t及之前的window-1个数据不包含未来数据所以这里没有未来函数问题。但volume_mean的使用需要注意窗口的对齐方式rolling(window).mean()在当前时间点使用的是当前时间点及过去window-1个数据所以也不包含未来数据。5.3 计算持续性图点云构造好之后调用ripser计算持续性图。ripser的输入是二维数组每一行是一个点列是坐标维度。它返回一个包含不同维度的持续同调信息的列表dgms[0]是H0的结果dgms[1]是H1的结果。# 文件路径src/tda_factor.py from ripser import ripser def compute_persistence(X, maxdim1): result ripser(X, maxdimmaxdim) return result[dgms]ripser返回的H1矩阵里每一行是一组(birth, death)。但注意并不是所有洞都会在有限阈值内死亡。有些洞在过滤结束时仍然存在对应death是inf。在计算因子时要区分处理这些“未死亡”的结构可以直接丢弃也可以把inf替换为点云中所有点对距离的最大值。具体做法在后面的完整代码里演示。5.4 从持续性图提取因子有了持续性图矩阵之后因子提取就是把矩阵压缩成一个或几个标量。这里的关键是不要只取一个指标而是设计一组覆盖不同结构角度的指标。常用指标组合如下H1洞数量持续性图中有效洞的总数。H1总生命跨度所有洞里(death - birth)之和越大说明震荡结构越强。H1最大生命跨度最显著的那个结构持续多久。H1出生时间均值洞出现的尺度位置。H1死亡时间均值洞消失的尺度位置。H1生命跨度偏度洞的生命跨度分布是否偏向长尾。这组指标可以合成为一个因子向量后续既可以单独使用也可以作为机器学习模型的输入特征。5.5 滑动窗口与特征序列因子必须在时间序列上连续生成所以要采用滑动窗口方式。对每个窗口重复步骤5.2到5.4得到该窗口的因子值然后把因子值与窗口末尾的时间戳对齐。这样每个时间点都有一个由过去N根K线计算出的拓扑因子。窗口大小和滑动步长是TDA因子中最关键的超参数。窗口太小点云数量太少拓扑结构不稳定窗口太大Rips复形的计算量显著增加而且行情结构已经被平均化失去局部意义。从实践角度看5分钟K线或1分钟K线场景窗口取30到90比较合理日线场景窗口取60到120比较合理。具体数值需要结合标的和周期做参数扫描验证。6. Rips复形的复杂度窗口为什么窗口不能无限大有一个问题很多人会忽略Rips复形的计算复杂度非常高。假设点云里有N个点Rips过滤会枚举所有点对、所有三角形甚至更高维的单形。点对数量是O(N^2)三角形数量是O(N^3)如果计算到维度d复杂度接近O(N^(d1))。这个增长非常快。因此在实际计算中窗口大小不能随意设置。一个窗口取30个点点对只有435个取100个点点对就有4950个而三角形数量会达到16万级别。当窗口扩大到200个点时点对约2万个三角形约131万个计算耗时上升非常明显。解决这个问题的常用办法有三种控制窗口大小把窗口控制在30到60个数据点这是最直接的办法。降采样窗口内原始K线数量很多时先做等间隔降采样减少点云数量。比如原来1000根1分钟K线每5根取1根得到200个点再计算TDA。维度削减点云维度不是越高越好。经验上3到6维就能保留足够结构信息再高的维度只会增加计算开销对因子增量信息贡献有限。理解这个复杂度问题对后面实际跑数据非常有帮助。如果你发现代码运行很慢第一反应不是优化代码而是检查窗口大小和点云维度。7. 完整示例TDA行情结构因子挖掘代码下面给出一个完整的、可以直接运行的示例。先给出一个纯Python暴力计算版本帮你理解TDA内部发生了什么然后给出基于ripser的生产版本用于实际因子计算。7.1 纯Python暴力版理解TDA的边界这个版本不依赖TDA库只用itertools和numpy计算三个点形成的最小外接圆半径再统计所有三角形的半径分布。虽然这个版本不能计算完整的持续性图但能让你直观理解“洞”是怎么随距离阈值出现和消失的。# 文件路径demo/tda_bruteforce.py import itertools import numpy as np def circum_radius(p, q, r): 计算三个点的外接圆半径如果三点共线则返回最大值的一半 a np.linalg.norm(p - q) b np.linalg.norm(q - r) c np.linalg.norm(r - p) s (a b c) / 2.0 area_sq max(s * (s - a) * (s - b) * (s - c), 0.0) denom 2.0 * np.sqrt(area_sq) if denom 1e-12: return max(a, b, c) / 2.0 return (a * b * c) / denom def brute_force_tda(points): 暴力枚举所有三角形返回外接圆半径分布作为结构特征 triangles [] n len(points) for comb in itertools.combinations(range(n), 3): p points[comb[0]] q points[comb[1]] r points[comb[2]] radius circum_radius(p, q, r) triangles.append(radius) triangles np.array(triangles) return { triangle_count: len(triangles), mean_circum_radius: triangles.mean() if len(triangles) else 0.0, std_circum_radius: triangles.std() if len(triangles) else 0.0, max_circum_radius: triangles.max() if len(triangles) else 0.0, } if __name__ __main__: rng np.random.default_rng(42) # 生成一组接近圆环的点云模拟箱体震荡 theta np.linspace(0, 2 * np.pi, 30) circle np.column_stack([np.cos(theta), np.sin(theta)]) circle rng.normal(0, 0.03, sizecircle.shape) feats brute_force_tda(circle) print(feats)这个版本的输出是三个点形成的外接圆半径统计量。箱体行情中外接圆半径分布集中趋势行情中点云沿一条线展开外接圆半径会比较分散。这个直觉就是TDA结构因子的雏形。7.2 基于ripser的生产版本生产版本的核心是TDAFactor类它封装了数据加载、点云构建、持续性图计算和因子提取四个流程。# 文件路径src/tda_factor.py import numpy as np import pandas as pd from ripser import ripser from sklearn.preprocessing import StandardScaler class TDAFactor: def __init__(self, window60, stride5, maxdim1, scaleTrue): self.window window self.stride stride self.maxdim maxdim self.scale scale def _build_point_cloud(self, df): close df[close].values.astype(float) high df[high].values.astype(float) low df[low].values.astype(float) volume df[volume].values.astype(float) close_ma pd.Series(close).rolling(self.window).mean().values volume_mean pd.Series(volume).rolling(self.window).mean().values volume_std pd.Series(volume).rolling(self.window).std().values rolling_max pd.Series(high).rolling(self.window).max().values rolling_min pd.Series(low).rolling(self.window).min().values feat1 (close - close_ma) / np.where(np.abs(close_ma) 1e-12, 1e-12, close_ma) feat2 high - low feat3 (volume - volume_mean) / np.where(np.abs(volume_std) 1e-12, 1e-12, volume_std) feat4 np.append([0.0], np.diff(close) / np.where(np.abs(close[:-1]) 1e-12, 1e-12, close[:-1])) feat5 (close - rolling_min) / np.where( np.abs(rolling_max - rolling_min) 1e-12, 1e-12, rolling_max - rolling_min ) X np.column_stack([feat1, feat2, feat3, feat4, feat5]) # 去掉滚动窗口未形成的前几行 X[:self.window - 1] np.nan return X def _persistence_features(self, X): result ripser(X, maxdimself.maxdim) dgms result[dgms] h0 dgms[0] if len(dgms) 0 else np.empty((0, 2)) h1 dgms[1] if len(dgms) 1 else np.empty((0, 2)) # 过滤 inf 死亡点用点云最大成对距离替代 finite_h1 h1[np.isfinite(h1[:, 1])] if len(finite_h1) 0: finite_h1 np.empty((0, 2)) if len(finite_h1) 0: lifespans_h1 finite_h1[:, 1] - finite_h1[:, 0] total_life_h1 float(lifespans_h1.sum()) max_life_h1 float(lifespans_h1.max()) mean_life_h1 float(lifespans_h1.mean()) std_life_h1 float(lifespans_h1.std()) birth_mean_h1 float(finite_h1[:, 0].mean()) death_mean_h1 float(finite_h1[:, 1].mean()) else: total_life_h1 0.0 max_life_h1 0.0 mean_life_h1 0.0 std_life_h1 0.0 birth_mean_h1 0.0 death_mean_h1 0.0 finite_h0 h0[np.isfinite(h0[:, 1])] if len(finite_h0) 0: lifespans_h0 finite_h0[:, 1] - finite_h0[:, 0] total_life_h0 float(lifespans_h0.sum()) else: total_life_h0 0.0 return { h1_count: len(finite_h1), h1_total_life: total_life_h1, h1_max_life: max_life_h1, h1_mean_life: mean_life_h1, h1_std_life: std_life_h1, h1_birth_mean: birth_mean_h1, h1_death_mean: death_mean_h1, h0_total_life: total_life_h0 } def extract(self, df): X self._build_point_cloud(df) n len(df) records [] start self.window - 1 for i in range(start, n - self.window 1, self.stride): window_X X[i - self.window 1: i 1] if np.any(np.isnan(window_X)): continue feats self._persistence_features(window_X) feats[datetime] df[datetime].iloc[i] feats[close] df[close].iloc[i] records.append(feats) feature_df pd.DataFrame(records) return feature_df这段代码里比较关键的设计有三个一是滚动窗口的特征标准化。这里的_build_point_cloud使用滚动窗口计算MA和标准差而不是全局统计量降低未来函数风险。需要说明的是为了演示简洁这里没有对特征做归一化到统一尺度实际项目建议在窗口内部再做一次标准归一化。二是inf死亡点的处理。ripser在过滤结束时可能仍有洞未死亡返回deathinf。这里把它们过滤掉只统计有限生命跨度的结构。如果某个窗口所有洞都是无穷生命跨度那就说明该窗口点云没有形成任何闭合空间这本身也是有效信息。三是滑动窗口的对齐。使用stride控制计算频率时间戳对齐到窗口最后一个K线。这样可以控制因子计算频率也方便合并到下游策略。7.3 主程序运行主程序做的事情是读取数据创建TDAFactor对象调用extract方法得到因子序列并保存到CSV文件。# 文件路径scripts/run_tda_factor.py import pandas as pd from src.tda_factor import TDAFactor df pd.read_csv(data/stock_data.csv, parse_dates[datetime]) df df.sort_values(datetime).drop_duplicates(subsetdatetime).reset_index(dropTrue) # 如果数据不是按时间升序先排序 tda TDAFactor(window60, stride5, maxdim1) feature_df tda.extract(df) print(feature_df.head()) print(feature_df.describe()) output_path output/tda_factor.csv feature_df.to_csv(output_path, indexFalse) print(ffactor saved to {output_path})运行命令cd tda_quant python scripts/run_tda_factor.py预期会看到类似下面的输出列名一致具体数值取决于数据datetime close h1_count h1_total_life ... 0 2024-01-08 10:30:00 3021.50 2 0.312 ... 1 2024-01-08 10:35:00 3025.80 3 0.458 ... ...这说明每个时间点都对应一组拓扑结构因子。接下来要做的是因子验证。8. 运行结果与效果验证怎么判断因子有没有价值构造完因子后第一件事不是直接扔进回测而是做三类验证结构合理性、相关性检验、预测能力检验。8.1 结构合理性检查先看因子的时间序列分布是否符合直觉。可以绘制因子值和行情K线在同一张图上观察箱体震荡区间H1洞数量是否偏高、趋势段H1洞数量是否偏低。这个观察性检查能快速发现代码逻辑错误。import matplotlib.pyplot as plt fig, ax1 plt.subplots(figsize(12, 6)) ax1.plot(df[datetime], df[close], colorblack, alpha0.7, labelclose) ax2 ax1.twinx() ax2.plot(pd.to_datetime(feature_df[datetime]), feature_df[h1_count], colorred, alpha0.7, labelh1_count) plt.title(TDA Factor H1 Count vs Price) plt.legend() plt.show()如果发现趋势行情里H1数量反而很高那大概率是点云构造方式有问题比如特征维度选择不当导致距离度量失真。8.2 与已有因子的相关性检验把TDA因子和已有传统因子放在一起计算相关性矩阵。理想情况下TDA因子与传统动量、波动率因子的相关系数应该在0.2以下与成交量类因子的相关系数也应当较低。如果相关系数过高说明TDA因子捕获的仍然是传统信息增量有限。8.3 预测能力检验一个常用的轻量级验证方法是计算因子值与下一期收益的秩相关系数IC。具体做法是把特征按时间顺序切片下一期收益用未来N根K线的累计收益计算Spearman秩相关。import numpy as np from scipy import stats feature_df feature_df.sort_values(datetime).reset_index(dropTrue) close_map pd.Series(df[close].values, indexpd.to_datetime(df[datetime])).sort_index() aligned_close close_map.reindex(pd.to_datetime(feature_df[datetime])).values future_return pd.Series(aligned_close).shift(-5) / aligned_close - 1.0 valid future_return.notna() ic, p_value stats.spearmanr(feature_df[h1_total_life][valid], future_return[valid]) print(fIC: {ic:.4f}, p-value: {p_value:.4f})注意这里的IC计算只是示例真正做因子研究时需要更严谨的分组检验、分层回测和样本外验证不能因为单段IC高就认定因子有效。9. 常见问题与排查方法我在写TDA因子代码时遇到过几个比较典型的坑整理成排查表供你对照问题现象可能原因排查方式解决方案ripser运行非常慢窗口过大点云数量过多统计点云规模测试不同窗口耗时降采样或减小窗口点云维度控制在6维以内返回的H1矩阵为空点云形状简单没有形成闭合洞检查点云是否接近一条直线或集中在极小区域增加点云复杂度调整特征构造大量deathinf过滤参数达到最大值时洞仍未闭合查看点云最大成对距离过滤inf或用最大成对距离替代特征值出现NaN滚动窗口早期数据不足或分母为0打印特征矩阵检查rolling对齐丢弃窗口未形成的行分母加微小量因子值与行情明显不符点云特征标准化方式错误对照K线图观察检查是否错误地使用了全样本标准化IC方向不稳定窗口参数不合适或特征构造不稳定做参数敏感性分析使用多个窗口做因子合成而非单窗口因子与传统因子高度相关点云特征被某个维度主导检查特征维度方差贡献增加维度或改用更平衡的特征组合其中最常见的还是第一个计算慢。Rips复形虽然是目前计算持续同调的高效算法但它仍然需要枚举大量单形点云规模稍微上去计算开销就会指数增长。生产环境下建议先用一个小数据子集跑通流程确认因子逻辑正确后再考虑用并行或抽样方式扩大计算规模。10. 最佳实践与工程建议这部分是实践经验的总结。如果要把TDA因子真正用到研究或生产环境下面几条建议值得认真考虑。10.1 参数不要一次定死窗口大小、滑动步长、点云维度、特征选择这些参数对因子表现影响很大。建议设计一组参数网格做敏感性分析看因子IC的均值和稳定性。如果因子IC只有在某一组精确参数下才有效在其他参数下全部失效那这个因子大概率是过拟合的。理想情况是在参数网格的一个较广范围内因子方向一致IC保持在合理水平。10.2 警惕未来函数TDA计算中最大的未来函数风险不在ripser本身而在特征构造。比如用全局的均值和标准差做标准化就会把未来数据泄露到当前特征中。正确的做法是用滚动窗口计算或者先对数据集做时间切分在训练集上拟合标准化参数再应用到测试集。回测系统里尤其要注意这一点。10.3 因子入库前要做完整校验如果把因子落库到因子库或数据库必须遵循合法授权、测试环境验证、备份、回滚和最小权限原则。具体来说先在测试库执行建表和写入脚本确认字段类型、空值比例、时间对齐都正确生产库执行前备份相关表写入因子数据时只使用最小权限账号禁用DROP、TRUNCATE等高危操作。因子数据一旦污染影响的是后续所有策略所以宁可慢一点也不能跳过验证。10.4 计算TDA特征时的资源控制TDA特征计算属于计算密集型任务不适合在主线程回测循环里直接调用。建议在因子预处理阶段一次性计算并保存然后回测时读取缓存。如果标的多、频率高可以按标的并行计算或者用采样点云近似计算不必对每一根K线都做完整TDA。10.5 多窗口合成比单窗口更稳健单窗口TDA特征容易受噪声和参数影响。一种更稳的做法是同时计算多个窗口比如30、60、120的TDA特征然后把同一类特征做标准化后取平均。这类似多尺度特征融合能让因子保留不同级别行情结构的信息降低单窗口参数敏感性。10.6 别忽视可解释性TDA因子虽然来自数学概念但并不意味着必须黑箱使用。每一步都可以加可解释的约束比如窗口内点云可视化、持续性图可视化、结构切换事件与行情事件的对照。保留这些信息后续做因子归因和策略复盘时非常有价值。11. 总结与后续方向至此你应该已经理解TDA在行情结构因子挖掘中的定位它不是替代传统的动量、波动率、成交量因子而是补充一类传统统计量捕捉不到的“结构维度”信息。核心流程可以归结为四步构造点云、计算持续性图、提取拓扑特征、做因子验证。本文给出了从数据准备到因子落地的完整Python代码你在自己的数据上跑通之后可以把H1生命跨度、洞的数量等特征作为新的特征列加入现有因子库然后与传统因子做相关性分析看看真正增加了多少信息。后续可以深入的方向包括持续性图的高阶统计量比如持续熵、持续景观的L2范数把TDA特征作为图神经网络的节点特征以及多标的多周期的TDA因子合成。这些方向都在朝着同一个目标前进让机器能够更完整地理解行情的形状而不只是行情的数字。如果这篇文章对你有帮助建议收藏备用后续做因子挖掘时可以随时对照代码结构复用。有一个善意的提醒TDA因子的有效性高度依赖标的、周期和数据质量从一个市场得出的规律不一定能迁移到另一个市场。无论因子表现多好都要先在样本外、在严格的回测框架里验证再考虑实盘使用。量化研究里的每一步都要对风险保持足够的敬畏。