# DTW算法实战：用Python实现动态时间规整匹配股票走势（附完整代码） 最近和几位做量化研究的朋友聊天，大家不约而同地提到了一个痛点：怎么判断两支股票的价格走势“像不像”？直接拿日K线图叠在一起看，如果时间轴对不齐，或者一只股票上市早、一只上市晚，比较起来就非常别扭。这就像让两个身高、步幅都不一样的人赛跑，单纯比较某个时间点的位置毫无意义，你得想办法把他们“走路的过程”对齐了看。在时间序列分析领域，**动态时间规整**（Dynamic Time Warping, DTW）就是解决这类问题的“神兵利器”。它不要求两个序列长度相同，能弹性地拉伸或压缩时间轴，找到最佳匹配路径，从而计算出最真实的相似度。对于金融数据分析、量化策略中的模式识别，DTW提供了一种远比简单相关系数更细腻、更强大的视角。今天，我们就抛开复杂的数学推导，直接上手Python，从数据获取到可视化，完整走一遍用DTW分析股票走势相似度的实战流程。 ## 1. 理解核心：为什么DTW比传统方法更适合金融序列？ 在量化分析中，我们常使用相关系数或欧氏距离来衡量两个时间序列的相似性。但这些方法有一个致命假设：**序列必须是等长且时间点严格对齐的**。股票市场里，这种理想情况几乎不存在。 * **上市时间不同**：A股2010年上市，B股2015年上市，前5年的数据直接缺失。 * **走势相位偏移**：板块轮动中，龙头股可能先启动，跟风股晚几天反应，但趋势形态相似。 * **波动速率差异**：一只股票可能在3天内完成一波急涨，另一只则用10天缓慢爬升，形成类似的“圆弧底”。 这时，**欧氏距离**会因时间点错位而计算出极大的距离，认为两者毫不相似。**DTW**则通过允许序列在时间轴上进行非线性的“规整”，找到使累积距离最小的对齐方式。它的核心思想是：**我们关心的是形状的相似，而非严格的时间点对应**。 > 提示：你可以把DTW想象成给两条序列做“动态规划下的最佳配对”。它要找到一条路径，让序列A的每个点都能在序列B中找到“最合适”的对应点（可以一对多），使得所有配对点的局部距离之和最小。 为了更直观地对比，我们看一个简单例子。假设我们有两个序列： ```python import numpy as np series_A = np.array([1, 3, 4, 6, 7, 8]) # 序列A series_B = np.array([2, 3, 5, 7]) # 序列B，更短 ``` 计算它们的欧氏距离（需要插值或截断对齐，这里假设截断），与DTW距离的感受完全不同。DTW会允许`series_A`中的`[4,6]`与`series_B`中的`5`匹配，从而捕捉到中间隆起的共同形态。 下表快速对比了几种常见相似度度量方法在金融序列分析中的表现： | 度量方法 | 是否需要等长 | 是否考虑时间扭曲 | 对相位偏移敏感度 | 计算复杂度 | 适用场景 | | :--- | :--- | :--- | :--- | :--- | :--- | | **皮尔逊相关系数** | 是 | 否 | 极高 | O(n) | 同步性高的序列，如同一时段内大盘与ETF | | **欧氏距离** | 是 | 否 | 极高 | O(n) | 严格对齐的标准化后序列比较 | | **余弦相似度** | 是 | 否 | 高 | O(n) | 忽略幅度、只关注方向变化的分析 | | **DTW距离** | **否** | **是** | **低** | O(n*m) | **走势形态匹配、异步序列对比、模式识别** | 显然，当我们的目标是寻找“走势相似”的股票时，DTW的优势是压倒性的。尽管其计算复杂度更高（O(n*m)），但对于日常的股票数据分析（单支股票几千个交易日数据），现代计算机完全能轻松应对。 ## 2. 实战准备：搭建Python环境与获取股票数据 理论说得再多，不如一行代码。我们首先确保环境就绪。推荐使用`Anaconda`创建独立的Python环境，避免包版本冲突。 ```bash # 创建并激活一个名为`dtw_stock`的虚拟环境（可选但推荐） conda create -n dtw_stock python=3.9 conda activate dtw_stock # 安装核心库 pip install numpy pandas matplotlib yfinance scikit-learn fastdtw ``` 这里我们引入了几个关键库： * `yfinance`：从雅虎财经免费获取股票历史数据，非常方便。 * `fastdtw`：一个高效的DTW算法实现库，比纯Python循环快几个数量级。 * `scikit-learn`：用于数据标准化等预处理。 接下来，我们获取两支大家熟悉的、业务可能有关联的股票数据进行比较。例如，我们比较**比亚迪（002594.SZ）** 和**宁德时代（300750.SZ）** 在2023年的股价走势。 ```python import yfinance as yf import pandas as pd import matplotlib.pyplot as plt # 定义股票代码和时间范围 tickers = ["002594.SZ", "300750.SZ"] start_date = "2023-01-01" end_date = "2023-12-31" # 下载数据 data = yf.download(tickers, start=start_date, end=end_date, progress=False) # 我们使用调整后的收盘价（Adjusted Close），它考虑了分红、拆股等因素 price_df = data['Adj Close'].dropna() # 删除缺失值 print(price_df.head()) print(f"\n数据形状: {price_df.shape}") ``` 运行后，你会得到一个`DataFrame`，索引是交易日，两列分别是两支股票的复权价格。如果数据获取成功，就可以进入下一步。 > 注意：`yfinance`的代码后缀需与市场对应（.SZ深交所，.SH上交所）。网络请求可能受地域或时效影响，如果失败，可以尝试更换时间段，或使用`akshare`等国内替代库。 ## 3. 核心步骤：数据预处理与DTW算法实现 原始股价数据不能直接用于DTW计算。比亚迪股价在200多元，宁德时代在100多元，绝对数值的差异会淹没形态信息。我们需要进行**标准化**。 ### 3.1 数据标准化与序列提取 我们通常对价格序列计算**日收益率**，或者进行**Z-Score标准化**。这里使用收益率，它能更好地反映价格变化的相对形态，且是金融分析中的标准做法。 ```python # 计算日收益率序列 returns_df = price_df.pct_change().dropna() # 提取为numpy数组 series_byd = returns_df["002594.SZ"].values series_catl = returns_df["300750.SZ"].values print(f"比亚迪收益率序列长度: {len(series_byd)}") print(f"宁德时代收益率序列长度: {len(series_catl)}") # 理论上此时两者长度应相同，因为剔除了同期的NaN ``` ### 3.2 亲手实现一个基础的DTW算法 虽然`fastdtw`很好用，但理解其原理至关重要。下面我们实现一个标准的、基于动态规划的DTW算法，它清晰地展示了距离矩阵和最优路径的寻找过程。 ```python import numpy as np def dtw_distance_manual(x, y): """ 计算两个序列x和y之间的DTW距离。 返回：累计距离矩阵，DTW距离，最优路径 """ n, m = len(x), len(y) # 初始化距离矩阵，用无穷大填充 dtw_matrix = np.full((n+1, m+1), np.inf) dtw_matrix[0, 0] = 0 # 计算局部距离矩阵（欧氏距离） for i in range(1, n+1): for j in range(1, m+1): cost = abs(x[i-1] - y[j-1]) # 局部成本，这里用绝对值差，对于收益率也可用平方差 # 动态规划递推：当前最小累积成本 = 局部成本 + 之前三个方向的最小成本 last_min = min(dtw_matrix[i-1, j], # 插入 dtw_matrix[i, j-1], # 删除 dtw_matrix[i-1, j-1]) # 匹配 dtw_matrix[i, j] = cost + last_min # 回溯寻找最优路径 path = [] i, j = n, m while i > 0 or j > 0: path.append((i-1, j-1)) # 记录到路径中（转换为0-based索引） if i == 0: j -= 1 elif j == 0: i -= 1 else: min_prev = min(dtw_matrix[i-1, j], dtw_matrix[i, j-1], dtw_matrix[i-1, j-1]) if min_prev == dtw_matrix[i-1, j-1]: i -= 1 j -= 1 elif min_prev == dtw_matrix[i-1, j]: i -= 1 else: # dtw_matrix[i, j-1] j -= 1 path.reverse() return dtw_matrix, dtw_matrix[n, m], path ``` 这个函数是DTW的核心。它创建了一个`(n+1)*(m+1)`的矩阵，通过动态规划填充每个单元格的**最小累积距离**。回溯路径则告诉我们，两个序列的各个点是如何最佳匹配的。 ### 3.3 使用优化库进行快速计算 对于实际分析，我们使用`fastdtw`，它采用了限制搜索范围的启发式方法，速度极快。 ```python from fastdtw import fastdtw from scipy.spatial.distance import euclidean # 使用fastdtw计算距离和路径 distance, path = fastdtw(series_byd, series_catl, dist=euclidean) print(f"比亚迪与宁德时代2023年日收益率序列的DTW距离为: {distance:.6f}") ``` 这个`distance`值就是衡量两者走势差异的量化指标。**值越小，说明在允许时间规整的前提下，两条序列的形态越相似**。 ## 4. 可视化：让DTW路径与序列比对一目了然 数字不够直观，我们需要图形来揭示DTW到底做了什么。我们将绘制三个子图： 1. 两支股票的原始收益率序列。 2. 对齐后的序列（通过路径映射）。 3. 距离矩阵与最优路径的热图。 ```python def plot_dtw_results(x, y, path, dtw_dist): fig = plt.figure(figsize=(15, 5)) # 子图1：原始序列 ax1 = plt.subplot(1, 3, 1) ax1.plot(x, label='BYD Returns', alpha=0.7) ax1.plot(y, label='CATL Returns', alpha=0.7) ax1.set_title('Original Return Series') ax1.set_xlabel('Trading Day Index') ax1.set_ylabel('Daily Return') ax1.legend() ax1.grid(True, linestyle='--', alpha=0.5) # 子图2：规整对齐后的序列 ax2 = plt.subplot(1, 3, 2) # 根据路径，获取对齐后的索引 x_aligned = [x[i] for i, _ in path] y_aligned = [y[j] for _, j in path] time_axis = np.arange(len(path)) ax2.plot(time_axis, x_aligned, label='BYD (Aligned)', alpha=0.7) ax2.plot(time_axis, y_aligned, label='CATL (Aligned)', alpha=0.7) ax2.set_title(f'Aligned Series (DTW Distance: {dtw_dist:.2f})') ax2.set_xlabel('Warped Time Index') ax2.set_ylabel('Daily Return') ax2.legend() ax2.grid(True, linestyle='--', alpha=0.5) # 子图3：距离矩阵与最优路径（为了清晰，展示下采样后的矩阵） ax3 = plt.subplot(1, 3, 3) # 计算一个简化的距离矩阵用于展示（全矩阵太大） sample_step = max(len(x)//50, len(y)//50, 1) x_idx = range(0, len(x), sample_step) y_idx = range(0, len(y), sample_step) sample_matrix = np.zeros((len(x_idx), len(y_idx))) for i_idx, i in enumerate(x_idx): for j_idx, j in enumerate(y_idx): sample_matrix[i_idx, j_idx] = abs(x[i] - y[j]) # 绘制热图 im = ax3.imshow(sample_matrix.T, origin='lower', aspect='auto', cmap='hot_r') plt.colorbar(im, ax=ax3, label='Local Distance') # 绘制最优路径（映射到采样后的索引上） path_x = [i//sample_step for i, _ in path if i%sample_step==0 and i//sample_step < len(x_idx)] path_y = [j//sample_step for _, j in path if j%sample_step==0 and j//sample_step < len(y_idx)] # 确保长度一致 min_len = min(len(path_x), len(path_y)) if min_len > 0: ax3.plot(path_x[:min_len], path_y[:min_len], color='cyan', linewidth=1.5, label='Optimal Path') ax3.set_title('Distance Matrix & Warping Path') ax3.set_xlabel('BYD Index (sampled)') ax3.set_ylabel('CATL Index (sampled)') ax3.legend() plt.tight_layout() plt.show() # 调用绘图函数，使用fastdtw的结果 plot_dtw_results(series_byd, series_catl, path, distance) ``` 运行这段代码，你会得到一张信息丰富的图。在**对齐序列图**中，你可以看到两条收益率曲线被“拉”到了同一个时间轴上，波峰和波谷的对齐关系清晰可见。**热图中的青色路径**则直观展示了DTW是如何在距离矩阵中蜿蜒前进，寻找成本最低的匹配方式的。 ## 5. 进阶应用：在量化策略中批量寻找相似股票 单次比较只是开始。真正的威力在于批量处理。假设我们有一个股票池，想找出与“目标股票”走势最相似的N只股票。 ```python import warnings warnings.filterwarnings('ignore') def find_similar_stocks(target_ticker, candidate_tickers, start_date, end_date, top_n=5): """ 在候选股票列表中，找出与目标股票DTW距离最小的前N只。 """ # 1. 下载所有股票数据 all_tickers = [target_ticker] + candidate_tickers try: data = yf.download(all_tickers, start=start_date, end=end_date, progress=False)['Adj Close'] except Exception as e: print(f"数据下载失败: {e}") return None data = data.dropna(how='all', axis=1).dropna() # 清理数据 # 检查目标股票是否存在 if target_ticker not in data.columns: print(f"目标股票 {target_ticker} 在数据中不存在。") return None # 2. 计算收益率 returns_data = data.pct_change().dropna() target_series = returns_data[target_ticker].values # 3. 计算每只候选股票与目标股票的DTW距离 results = [] for ticker in candidate_tickers: if ticker not in returns_data.columns: continue candidate_series = returns_data[ticker].values try: distance, _ = fastdtw(target_series, candidate_series, dist=euclidean) results.append((ticker, distance)) except Exception as e: print(f"计算 {ticker} 时出错: {e}") continue # 4. 按距离排序，返回前N个 results.sort(key=lambda x: x[1]) return results[:top_n] # 示例：寻找与“贵州茅台”走势相似的消费类股票 target = "600519.SS" # 贵州茅台 candidates = ["000858.SZ", "600887.SS", "002304.SZ", "000568.SZ", "603288.SS"] # 五粮液、伊利、洋河、泸州老窖、海天味业 start = "2023-01-01" end = "2023-12-31" similar_stocks = find_similar_stocks(target, candidates, start, end, top_n=3) if similar_stocks: print("与目标股票最相似的前3只股票：") for ticker, dist in similar_stocks: print(f" 股票代码: {ticker}, DTW距离: {dist:.4f}") ``` 这个函数构建了一个简单的股票相似性扫描器。在实际策略中，你可以将其用于： * **配对交易**：寻找历史走势高度同步的股票对，当它们的DTW距离短期偏离历史均值时，可能蕴含套利机会。 * **行业板块分析**：计算板块内所有股票与板块指数的DTW距离，识别哪些个股是“板块纯正股”，哪些是“异动股”。 * **模式匹配**：将特定的价格形态（如V型反转、头肩底）作为目标序列，在全市场扫描出现类似形态的股票。 ## 6. 注意事项、局限性与性能优化 DTW并非银弹，在金融应用中需清醒认识其局限。 * **计算成本**：标准DTW复杂度为O(n*m)。对于长序列（如高频数据），计算压力大。 * **优化方法**：使用`fastdtw`库；设置**窗口约束**（如Sakoe-Chiba Band, Itakura Parallelogram），限制路径偏离对角线太远，这既符合金融序列的“大致同步”假设，又能将复杂度降至O(n*w)，其中w为窗口宽度。 ```python from fastdtw import fastdtw radius = 5 # 窗口半径 distance, path = fastdtw(series_byd, series_catl, radius=radius, dist=euclidean) ``` * **距离的对称性与量纲**：DTW距离不是严格意义上的度量，有时不满足三角不等式。且距离值受序列量纲影响。**务必在相同标准化的序列间进行比较**。 * **过拟合风险**：DTW的弹性很强，甚至能让两条随机序列产生一个看似很低的距离。**一定要结合可视化判断**，看看最优路径是否合理，对齐后的形态是否真的相似。 * **与弗雷歇距离的区别**：输入中提到了弗雷歇距离（Fréchet distance）。它常被比作“遛狗距离”，关心的是两条曲线在保持各自点序情况下的最大 leash 长度，对整体形状的匹配要求更“严格”，计算也更复杂。DTW则更关注累积成本，对局部拉伸更宽容。在股票分析中，DTW通常更实用。 我在实际构建选股因子时，曾将DTW距离与波动率、成交量指标结合，形成了一个“形态稳定性因子”。初期没有加窗口约束，结果发现一些涨跌周期完全错位的股票也被判为相似，产生了不少噪音交易。后来加上时间窗口约束，并**要求最优路径的斜率大部分时间接近1**，因子的有效性才显著提升。另一个坑是数据频率，用日线数据和用周线数据计算出的相似度排名可能差异很大，这取决于你的策略交易周期。对于中长期持仓，用周线或月线数据计算DTW，或许能过滤掉日线级别的噪音，找到真正趋势一致的标的。