用Python做股票价格序列相似性分析：DTW与形态匹配实践

简介：基于Python的股票价格序列相似性分析课程设计资源包，面向金融数据分析、Python编程及算法课程设计人群，围绕动态时间弯曲（DTW）算法实现股票价格序列的相似性度量，并通过折线图直观呈现对比结果，解决同类课程中序列对齐与形态比较的典型问题。压缩包共12个文件，核心为2个Python源码（Main.py、source.py）与1份Word版课程设计报告，另含requirements.txt依赖说明、database.db数据文件及7张运行效果截图，整体大小仅2.13MB，结构紧凑便于对照学习。平台显示已有613人学习下载，认可度良好。资源提供了从数据读取、数据预处理到DTW相似性计算、结果可视化的完整代码流程，配合报告中的算法原理讲解与截图展示，可帮助读者快速理解序列相似性分析的实现思路，并直接用于相关课程设计、期末项目或作为后续研究参考。

1. 用 Python 做股票价格序列相似性分析：这个 zip 解决的问题比想象中更具体

拿到这个 zip 的人，多半正在找一种能回答这类问题的代码：当前这 20 根、40 根 K 线，历史上有没有出现过差不多的形态？相似形态出现之后，行情是继续走、还是反转？“股票价格序列相似性分析”干的就是这件事——把当前一段行情切片，和历史所有切片做匹配，按相似度排序。它常见于形态复盘、历史行情回放，也是形态类量化策略的信号来源。Python 这条技术路径的好处是生态齐全，从数据获取到相似度计算再到可视化，都能在几十行代码里跑通。先说一个反直觉的结论：直接算欧氏距离判断两条 K 线像不像，在真实行情里会频繁翻车，原因是两个序列往往存在相位偏移，相似形态不一定对齐在相同的时间点上，真正靠谱的是 DTW 这类允许时间轴弯曲的度量方式。

2. 从数据到序列：价格数据获取与预处理的两个关键选择

2.1 获取前复权行情：免费数据源与复权参数怎么配

相似性分析的第一步是拿到干净、连续、可对齐的价格序列。常见做法是用 akshare 做快速验证，免费、不需要注册 token，接口返回的就是 pandas DataFrame，直接能用；如果团队内部已经有 tushare pro 的 token，也可以用ts.pro_bar拿同样的数据。下面以 akshare 为例，取平安银行 2020 年到 2024 年的日线：

import akshare as ak # symbol 传股票代码，period 固定为 daily，adjust="qfq" 表示前复权 df = ak.stock_zh_a_hist( symbol="000001", period="daily", start_date="20200101", end_date="20241231", adjust="qfq" ) # 只需要日期和收盘价，按时间升序排好并重建索引 df = df.rename(columns={"日期": "trade_date", "收盘": "close"}) df = df[["trade_date", "close"]].sort_values("trade_date").reset_index(drop=True) print(df.head())

adjust="qfq"是这里最关键的参数。除权除息日当天，股价会向下跳空，不复权数据里这个跳空会被相似性算法识别成一次剧烈的“下跌”，导致所有跨越除权日的切片距离被拉高，匹配结果里出现一批和形态毫无关系的假相似。前复权会按最新股本把历史价格整体调整，保持价格走势连续，是形态类分析用的默认设置。注意它的副作用：复权时间久了，早期价格可能被调整为负数，做收益率计算时要留个心眼，十年以上的长历史建议改用adjust="hfq"后复权。

另外两个容易忽略的细节：一是停牌日可能没有行情记录，读取后要检查 NaN，用ffill填充或直接剔除；二是在 Windows 上解压这个 zip 时，如果解压路径带中文，pandas 读 CSV 很容易撞上编码错误，把路径改成纯英文能省掉很多排查时间。运行代码前确保环境里已经装好 numpy、pandas 和 akshare，装 numpy 用pip install numpy，装不上时先检查 Python 是否在系统 PATH 里。

2.2 价格序列还是收益率序列：两种输入各自的含义和坑

拿到收盘价之后，下一个选择直接影响“相似”的定义。直接用价格序列算相似度，不同股票、不同历史时段的绝对价位差异会主导距离——茅台 800 元时期的走势和一只 8 元股票的走势，即使形态一模一样，价格尺度也差了 100 倍，必须做标准化。标准化价格序列保留趋势和幅度信息，适合回答“哪段走势长得像当前这段”。收益率序列则天然去趋势，适合回答“哪段走势的拐点节奏和现在一致”。下面两种预处理都写出来：

import pandas as pd import numpy as np def zscore(s: pd.Series) -> pd.Series: # z-score 标准化：去掉均值、除以标准差，消除绝对价位影响 return (s - s.mean()) / s.std() # 方式一：标准化价格序列，保留趋势形态 price_z = zscore(df["close"]) # 方式二：收益率序列，天然去趋势；pct_change 首行是 NaN，需要丢弃 ret = df["close"].pct_change().dropna()

我一般会把两种都算一遍，分两个 pipeline 跑，然后对比结果。原因是它们对同一段行情的敏感度完全不同：标准化价格对趋势敏感，一段横盘和一段稳步上涨的形态会拉开距离；收益率序列对单日极端波动敏感，一个涨跌停就能把整段序列的重心带偏，但它的好处是跨股票可比。如果你做的是全市场选股，想在不同股票之间找相似形态，优先用收益率序列；如果你做的是单只股票的历史复盘，想找“历史上我见过这种走法”，标准化价格更直观。

预处理阶段还有一个容易犯的错：窗口内的标准化必须在切片内部做，而不是对整个历史序列统一做标准化。原因是相似性比较看的是切片内的相对形态，如果直接用全序列的均值和标准差去归一化，不同历史时期的波动率水平会干扰匹配，早期的低波动切片会被错误放大。

3. 相似性度量怎么选：DTW 为什么在股票序列上优于欧氏距离

3.1 三种度量方法对比：距离定义差异决定了匹配质量

把序列切成片段之后，核心问题变成：怎么定义“像”。三种常用方法各自有不同的距离定义，直接决定了匹配效果。欧氏距离要求两个序列等长且逐点对齐，时间轴上一旦错位，距离就急剧放大；皮尔逊相关系数衡量线性相关，对幅度不敏感，但同样需要对齐，而且它只能捕捉同步的形态关系；动态时间规整（DTW）通过动态规划寻找两个序列之间的最优对齐路径，允许时间轴局部拉伸或压缩，天然适配股票数据的相位偏移问题。三类对比如下：

股票序列里的“相似”，本质上是形态相似，而不是数值恰好相同。当前 40 日 K 线可能和历史上某一个 60 日 K 线的中间段长得一样，只是当时走得慢、磨得久，欧氏距离要求头对头、尾对尾，错一天距离就被放大一倍。DTW 允许路径往左、往右、往对角线方向走，找到累计距离最小的对齐方式，形态相似的两段序列即使长度不同、节奏不同，也能被正确识别。代价是计算复杂度从 O(n) 涨到 O(n*m)，二十年的日线全历史暴力匹配会有性能压力，这个坑放在第五章细说。

3.2 用 numpy 写一个最小 DTW 实现：递推矩阵与路径约束

工程上可以直接调fastdtw库，但我还是建议手写一个最小实现，几十行代码能让你把 DTW 的递推逻辑彻底搞清楚，后面调参数、加约束、排查异常匹配都有底。核心是一个累积代价矩阵，每个格子的值是当前两个点的绝对差，加上到达这个格子之前最小的累计代价：

import numpy as np def dtw_distance(s1: np.ndarray, s2: np.ndarray, window: int = None) -> float: """ 最小 DTW 实现，返回两个序列的最小累计距离。 s1, s2: 一维序列，建议已经做过分段内标准化 window: 路径约束，限制 i 和 j 的偏移量，防止荒谬拉伸 """ n, m = len(s1), len(s2) # 初始化 n+1 x m+1 矩阵，边界值为正无穷，保证路径不从边界外绕行 dtw = np.full((n + 1, m + 1), np.inf) dtw[0, 0] = 0.0 for i in range(1, n + 1): for j in range(1, m + 1): # 如果设置了 window，偏移超过阈值的格子直接跳过 if window is not None and abs(i - j) > window: continue cost = abs(float(s1[i - 1]) - float(s2[j - 1])) # 递推式：当前点代价 + 到达前一个点的三条路径中的最小值 dtw[i, j] = cost + min( dtw[i - 1, j], # 向上走，s2 多对齐一次 dtw[i, j - 1], # 向右走，s1 多对齐一次 dtw[i - 1, j - 1] # 对角线走，两点正常对齐 ) return dtw[n, m]

递推式的几何意义很重要：dtw[i, j]是序列1 前 i 个点和序列2 前 j 个点的最小累计代价。向上走意味着序列2 的当前点重复对齐序列1 的当前点，向右走则反过来，这种“重复对齐”就是 DTW 处理时间轴扭曲的机制。window参数是血泪经验换来的——如果完全不限制路径，DTW 可能通过极端拉伸找出一条奇怪的路径，让两个完全不相关的片段也拿到很小的距离。经验值取序列长度的 10%，40 日窗口就设window=4。

还要注意返回值是累计代价，不是平均代价。两个序列越长，累计代价天然越大，跨长度比较时必须归一化，用dtw_distance(s1, s2) / (len(s1) + len(s2))换算成每个点的平均代价，否则匹配结果永远偏向短序列。

3.3 滑窗切分：把历史行情切成可比片段再匹配

有了距离函数，下一步是把历史行情切成一堆等长的片段，建立历史形态库。切分用滑窗，窗口长度决定你在找“多长”的形态，步长决定匹配密度。常见做法是窗口取 40 个交易日（约两个月），步长取 1 或 5，前者最细、计算量大，后者做粗筛。切分时每个片段内部做 z-score 标准化，这一步必须在切片内做，原因前面已经说过：

import numpy as np import pandas as pd def build_window_library(close: pd.Series, window: int = 40, step: int = 5): """ 把收盘价序列切成等长窗口，每个窗口内独立做 z-score。 返回窗口矩阵和每个窗口的截止日期索引。 """ arr = close.to_numpy() windows = [] end_dates = [] # 滑窗切分：python 数组切片，左闭右开，arr[start:start+window] for start in range(0, len(arr) - window + 1, step): seg = arr[start:start + window] # 每个窗口独立标准化，剔除绝对价位和波动率差异 seg_norm = (seg - seg.mean()) / (seg.std() + 1e-8) windows.append(seg_norm) end_dates.append(close.index[start + window - 1]) return np.vstack(windows), np.array(end_dates)

窗口长度是这个环节最该调的参数。20 日太短，噪声占比高，随便两段随机波动都可能算出低距离；60 日以上自然匹配的片段数量骤减，能选出来的 TopN 参考意义下降。我一般先跑 40 日窗口 + step=5 做第一轮粗筛，锁定相似度排名前 100 的片段，再把这些片段用原始逐日数据做一次精排，兼顾效率和精度。step=5会让相邻窗口高度重叠，匹配结果里经常出现同一段行情被连续多日命中，这是正常的，后续去重按截止日期做就行。

4. 从相似度到可复现结果：TopN 排序、可视化与信号转换

4.1 对全历史片段做 TopN 排序：归一化距离再比

历史形态库建好之后，拿当前最新的窗口去和库里所有历史窗口算距离，排序取 TopN。注意比较前必须做归一化，直接用原始 DTW 累计距离排序，结果会偏向短序列或低波动片段，这是新手最容易看错的一步。排序代码：

import numpy as np # 取最后一个窗口作为当前形态（假设当前是库的最后一段） cur = windows[-1] scores = [] # 只和历史窗口比，不和自己比；倒数第二个窗口和当前高度重叠，也建议跳过 for i in range(len(windows) - 10): d = dtw_distance(cur, windows[i], window=4) # 归一化：除以两个序列总长度，换算成平均每点最小代价 norm_d = d / (len(cur) + len(windows[i])) scores.append((end_dates[i], norm_d)) # 按归一化距离升序，取前 10 个最相似的时段 top10 = sorted(scores, key=lambda x: x[1])[:10] for date, score in top10: print(date, round(score, 4))

这里window=4对应 40 日窗口的 10% 偏移约束。如果信号是 20 日窗口，则改为window=2。排名结果中排名靠前的片段大概率集中在同一个历史牛熊阶段的相邻日期，因为行情本身有自相关性，此时可以做一次去重：如果两个命中日期间隔小于窗口长度，保留距离更小的那个。

4.2 可视化：当前走势和 Top3 历史走势并排对比

排名数字不足以说服自己或团队，可视化才是验证相似度计算是否合理的手段。画出当前序列和历史 Top3 序列的对比图，注意用同一个纵轴范围，否则标准化后的曲线尺度不一致，看起来会很误导：

import matplotlib.pyplot as plt fig, axes = plt.subplots(3, 1, figsize=(10, 8)) for idx, (date, score) in enumerate(top10[:3]): hist_start = np.where(end_dates == date)[0][0] axes[idx].plot(windows[-1], label="current", color="black") axes[idx].plot(windows[hist_start], label=str(date), color="steelblue", linestyle="--") axes[idx].set_ylim(-3, 3) axes[idx].legend() plt.tight_layout() plt.show()

视觉确认时要重点看拐点位置对不对。DTW 允许时间轴弯曲，意味着两条曲线高低点不必严格对齐，但拐点顺序应该一致——如果当前序列是“下台阶后横盘”，匹配到的历史片段也应该是“下台阶后横盘”，而不是“缓慢阴跌”或者“横盘后加速”。拐点对不上，说明窗口长度不合适，或者 window 约束太松。这种目视抽检每次至少做 3 个不同时间点的匹配，只看单次结果容易陷入“凑巧像”的错觉。

4.3 把相似变成信号：看匹配后的走势分布而不是单次结果

相似性分析本身只回答“像不像”，不回答“接下来涨不涨”。要用它辅助决策，标准做法是统计所有相似片段之后一段时间的收益分布，而不是盯住某一次匹配的后续走势。对每个命中的历史片段，取它结束之后未来 5 日的涨跌幅做分布统计：

def match_future_stats(close: pd.Series, windows: np.ndarray, end_dates: np.ndarray, top_k: int = 10, future: int = 5): """ 当前窗口匹配历史 TopK，统计每个历史片段未来 future 日的涨跌幅。 返回：平均涨跌幅、上涨比例、片段数量。 """ cur = windows[-1] scores = [] for i in range(len(windows) - 10): d = dtw_distance(cur, windows[i], window=4) scores.append((i, d / (len(cur) + len(windows[i])))) scores = sorted(scores, key=lambda x: x[1])[:top_k] rets = [] for i, _ in scores: start_pos = i * 5 # 窗口构建时 step=5，还原原始序列位置 end_pos = start_pos + 40 + future if end_pos >= len(close): continue seg = close.iloc[start_pos + 40: end_pos] rets.append(seg.iloc[-1] / seg.iloc[0] - 1) if not rets: return None return np.mean(rets), np.sum(np.array(rets) > 0) / len(rets), len(rets)

这段代码本质上是把“相似形态”翻译成“后续收益的样本分布”，已经接近一个形态类量化策略代码的雏形。但必须强调：统计结果只是描述，不是预测。匹配片段数量少于 30 次时，均值没有统计意义；涨跌幅分布要结合下一章的随机基准确认，否则很容易被幸存者偏差误导。

5. 股票序列相似性分析的避坑：5 条血泪经验里的现象、原因与修复

5.1 未复权数据在除权日制造“假相似”

现象：某只股票除权除息日之后，全历史相似度排名突然涌出一批和当前形态无关的片段；把命中片段画出来，发现它们的共同点是都跨过了一次除权跳空。

原因：除权日的价格跳空在 DTW 眼里是一个需要特殊拉伸才能对齐的突变点。为了绕开这个高成本区域，DTW 会倾向于把路径“跳过”这段，导致包含除权跳空的片段之间距离整体偏小，形成假相似。

解决：统一使用前复权数据，adjust="qfq"。如果历史区间超过十年，改用后复权避免早期价格为负。另外可以在预处理里加一道检查：相邻两日涨跌幅超过 15% 且当日成交量异常放大，优先怀疑是除权未处理，直接打印出来人工确认。

5.2 回测胜率虚高：历史形态库里混入了“未来数据”

现象：用相似形态做后续走势统计时，胜率高达 80% 以上，自己也觉得不真实；换了个时间段，胜率又掉到 40%，结果飘忽不定。

原因：构建历史形态库时没有做时序隔离。如果当前时间点是 T，而形态库里包含了 T 时刻之后才形成的片段，匹配时就会用到未来信息，这就是未来函数，回测曲线再漂亮也是假的。

解决：匹配第 T 日的形态时，候选库只保留截止日期早于 T 的片段，不能用全历史一次性建库再匹配。滚动更新库，每次把当前形态追加进库，同时对已有库按时间截断。判断一个代码有没有未来函数，看它建库和匹配是否在同一段循环里完成。

5.3 序列长度不一致时无脑补齐，距离被疯狂放大

现象：两个序列长度不同，比如一个是 40 日，一个是 55 日，直接拿 55 日序列的前 40 个点去比较，距离异常大；补零对齐后距离更大。

原因：补零相当于给序列两端人为制造了一段“零收益”，DTW 被迫用更长的路径绕过这段假数据，累计距离虚高，归一化后依然偏高。长度不一致时 DTW 的优势之一就是不用裁剪，但前提是参与比较的两个序列本身是“完整形态”，而不是硬截断的残片。

解决：做长度匹配时要么保持窗口严格等长，要么用 DTW 处理不等长序列后按平均代价归一化。等长窗口是工程上最省心的方案，构建历史形态库时就固定 window，后续所有比较都在同一长度下进行。

5.4 全市场暴力匹配太慢：从几十分钟到几秒的剪枝方法

现象：五千只股票、每只二十年日线、40 日窗口，直接全量跑 DTW，一次匹配要等十几分钟，完全没法迭代调参。

原因：手写双层循环本来就不快，再加上 Python 的 GIL 和动态类型开销，千万次级别的 DTW 调用就是灾难。DTW 本身的 O(n*m) 复杂度也扛不住全量扫描。

解决：先降复杂度再精排。第一步用皮尔逊相关系数粗筛，相关系数高于 0.8 的片段才进候选集，这一步是向量化的，几秒就能跑完；第二步只对候选集里的前 100 个片段做 DTW 精排。同时给 DTW 加 window 参数，把复杂度从 O(nm) 降到 O(nwindow)。数据量再大就上 numba JIT，或者用fastdtw的近似算法，精度损失在形态匹配场景下可以接受。

5.5 把“相似”当“因果”：忽略随机基准的比较

现象：某形态匹配完成后，后续 5 日平均收益 1.5%，看起来是一个可以入场的信号；但看一眼全样本，所有历史片段后续 5 日的平均收益也是 1.5%，根本没有超额。

原因：单次匹配的样本量小，偶然性高；很多“有效形态”只是刚好赶上了一段普涨行情。不做随机基准对比，任何统计结果都可能只是运气。

解决：每次统计相似片段后续收益时，从全部历史片段里随机抽同样数量做基准，对比两组均值差。样本量少于 50 次的匹配结论不要下；条件允许的话做 bootstrap 置信区间，看看均值差是否显著不为零。这是判断“相似性分析到底有没有用”的唯一可靠方式。

6. 进阶：DTW 窗口约束、多尺度确认与一个最小验证脚本

6.1 Sakoe-Chiba 带：给 DTW 加上合理的路径约束

DTW 的window参数就是 Sakoe-Chiba 带，限制路径偏移对角线的最大距离。经验值是序列长度的 10%：20 日窗口设 2，40 日窗口设 4，60 日窗口设 6。设置过小会把有意义的相位偏移全部截断，设置过大又回到无约束状态。调参时对比window=0和window=len*0.1的 Top10 结果，如果两组结果差异巨大，说明序列本身噪声过高，优先检查数据而不是继续调参。

6.2 多尺度确认：日线相似加上周线相似，结论才站得住

日线噪声会导致随机匹配，一种有效过滤方式是多尺度确认。日线匹配出 Top100 后，把命中的时间段换成周线再算一遍 DTW，取两组结果的交集。日线和周线都相似的形态，比单一日线相似可靠得多，相当于增加了一个独立验证维度。这个做法成本低，代码复用同一切片逻辑，只是把 resample 周期改成周线。

6.3 最小回测脚本：把最近 10 次形态匹配的后续走势跑出来

最后给一个可以直接用的回测骨架，验证相似度与未来收益之间是否存在真实关联。它把每个历史日期都当作“当前日”，滚动匹配，按平均相似度分高低两组，对比未来收益：

def similar_signal_backtest(close: pd.Series, window: int = 40, top_k: int = 10, future: int = 5): arr = close.to_numpy() n = len(arr) results = [] for t in range(window, n - future): # 用 t 日之前的 window 个点作为当前形态，不触碰未来数据 cur = arr[t - window:t] cur_z = (cur - cur.mean()) / (cur.std() + 1e-8) best = [] # 在更早的历史里匹配，步长 5 粗筛 for s in range(0, t - window + 1, 5): seg = arr[s:s + window] seg_z = (seg - seg.mean()) / (seg.std() + 1e-8) d = dtw_distance(cur_z, seg_z, window=4) / (2 * window) best.append((d, s)) if len(best) < top_k: continue best = sorted(best, key=lambda x: x[0])[:top_k] # 未来 future 日的实际涨跌幅，作为信号有效性标签 fwd_ret = arr[t + future] / arr[t] - 1 avg_sim = np.mean([x[0] for x in best]) results.append((fwd_ret, avg_sim)) results = np.array(results) median_sim = np.median(results[:, 1]) low_group = results[results[:, 1] < median_sim, 0] high_group = results[results[:, 1] >= median_sim, 0] return low_group.mean(), high_group.mean(), len(results) # 输出：(低相似组未来均值, 高相似组未来均值, 样本量) print(similar_signal_backtest(df["close"]))

注意这里的输出不构成交易建议，它只回答一个问题：历史相似度高低与未来收益之间有没有可观测的差异。如果高相似组的未来收益显著高于低相似组，说明这个形态窗口有信息量；如果没有差异，说明这个窗口长度或度量方式不适合当前标的，换参数重来。

我做这类分析最早的教训就是直接用 Top1 相似形态做预测，回测曲线画出来很漂亮，后来对比随机基线才发现根本没有优势。从那以后，我把这个工具的定位收敛成“形态定位器”——它负责告诉我历史上什么时候出现过类似走法，以及后来大概率的路径分布，最终判断还是交给人和更上层的规则。每次拿到新的股票代码，我会先跑一遍第六节的回测脚本看相似度有没有信息量，再决定要不要把参数调细。这个前置检查帮我避开了不少无效的形态研究，希望帮到你。

本文还有配套的精品资源，点击获取