1. 从“硬对齐”到“软匹配”：为什么金融时序预测需要DTW？

在金融世界里，我们每天都在和时间序列数据打交道。股票价格、交易量、汇率波动……这些数据点像一串串珍珠，被时间这根线串起来。新手朋友最常问的一个问题是：“我想找和某只股票走势相似的股票，或者预测它未来的走势，直接用欧几里得距离算一下差值不行吗？”

我刚开始做量化分析时，也是这么想的。结果踩了个大坑。当时我选了A股里两只同行业的龙头股，把过去一个月的日收盘价拉出来，直接用欧几里得距离算相似度。理论上，它们受同样的宏观环境影响，走势应该很接近。但计算结果却显示它们“很不相似”。我盯着K线图看了半天，明明肉眼看着趋势很像啊——都是先跌后涨，再盘整。问题出在哪呢？

后来我才明白，问题就出在“时间轴”的僵硬性上。欧几里得距离要求两个序列必须严格点对点对齐。这意味着，今天A股的价格必须和今天B股的价格比较，昨天对昨天，以此类推。但现实是，市场对信息的反应速度不同。同样一个利好消息，可能让A股在周一就大涨，而B股因为流动性或投资者结构的原因，延迟到周二甚至周三才启动。这就造成了时间轴上的错位或扭曲。一个上涨趋势，在A序列里被压缩在3天内完成，在B序列里可能缓慢地走了5天。用欧几里得距离去衡量，它们会因为相位不同而被误判为不相似。

这就像比较两个人唱同一首歌。一个人唱得快，一个人唱得慢，旋律和音调其实是一样的，但如果你硬要每一秒的声波都对上，那结果肯定是“完全不同”。DTW（动态时间规整）算法的高明之处就在于，它放弃了这种“硬对齐”，转而进行“软匹配”。它允许时间轴弹性伸缩，为序列A的某个点，在序列B中寻找一个最合适的对应点，而不是死板地要求时间戳必须一致。最终，它找到的是一条使两个序列整体形状最匹配的“对齐路径”。

在金融场景里，这种能力价值巨大。它可以帮我们：

• 识别相似模式：找到与目标股票走势高度相似的其他股票，用于构建配对交易策略。
• 预测走势：将当前走势与历史中的某段模式进行匹配，基于“历史会重演”的假设，对后续走势进行概率性推断。
• 检测市场状态：判断当前市场波动模式与历史上的“牛市启动”、“熊市反弹”或“盘整震荡”等典型状态的相似度。

简单说，DTW不是简单地比较数字，而是比较数字背后所代表的形态、趋势和节奏。它让计算机拥有了更接近人眼的、对图形模式的直觉判断力。

2. DTW核心原理：用动态规划给时间“做按摩”

理解了DTW的价值，我们来看看它到底是怎么工作的。抛开复杂的数学公式，我用一个最形象的比喻：DTW就像一个高级按摩师，它的任务是把两条僵硬的时间序列“按摩”到最贴合的状态。

假设我们有两个股价序列，序列X是 [1, 3, 4, 2]，序列Y是 [1, 2, 3, 4, 3]。它们长度不同，走势也有快慢。

第一步：计算“局部不适感” 按摩师首先要判断，把X的某个点和Y的某个点强行按在一起，会有多“别扭”。这个“别扭”程度就是局部距离。最常用的就是欧几里得距离的绝对值 d(i, j) = |X[i] - Y[j]|。我们会得到一个距离矩阵，里面每个格子 (i, j) 的值，就代表X的第i个点和Y的第j个点对齐时的成本。

第二步：规划“整体按摩路径” 按摩师不能只考虑一个点舒服，他要规划一条从起点（两个序列的开头）到终点（两个序列的结尾）的按摩路径，使得整条路径的“累计别扭程度”最小。这就是动态规划的核心思想。

他从起点 (1,1) 开始，思考走到任何一个格子 (i, j) 的最小累计成本。这个成本等于这个格子本身的“局部不适感”，加上从之前可能的位置过来的最小累计成本。通常，他只能从三个方向过来：左边 (i, j-1)（只拉伸Y）、上方 (i-1, j)（只拉伸X）、左上方 (i-1, j-1)（同时推进X和Y）。他每次都选成本最小的方向走。

用代码来看这个递推过程会更直观：

def dtw_basic(x, y):
    m, n = len(x), len(y)
    # 初始化一个巨大的累积成本矩阵
    dp = np.full((m, n), np.inf)
    dp[0, 0] = abs(x[0] - y[0])  # 起点成本

    # 填充第一行和第一列（边界，只能从一个方向来）
    for i in range(1, m):
        dp[i, 0] = abs(x[i] - y[0]) + dp[i-1, 0]
    for j in range(1, n):
        dp[0, j] = abs(x[0] - y[j]) + dp[0, j-1]

    # 动态规划填充整个矩阵
    for i in range(1, m):
        for j in range(1, n):
            cost = abs(x[i] - y[j])
            dp[i, j] = cost + min(dp[i-1, j],   # 从上方来
                                  dp[i, j-1],   # 从左方来
                                  dp[i-1, j-1]) # 从左上方来
    # 最终的最小累计成本就在右下角
    return dp[m-1, n-1], dp

最终，矩阵右下角 dp[m-1, n-1] 的值，就是完成整个“按摩”所需的最小总成本，也就是DTW距离。这个值越小，说明两个序列的形态越相似。

第三步：回溯“按摩手法” 得到最小成本后，我们从终点倒着走回去，看看按摩师具体是怎么走的。这条回溯的路径，就是最优的“对齐路径”。它明确告诉我们：序列X的第1个点，应该和序列Y的第1、2个点对齐（可能）；序列X中间某个快速上升的点，对应着序列Y里一段缓慢上升的区域。这条路径是DTW结果的灵魂，它揭示了两个序列在时间维度上是如何扭曲和匹配的。

3. 实战第一步：用DTW寻找走势相似的股票

理论讲得再多，不如亲手试一次。我们来做一个最经典的实战应用：从一堆股票里，找出和“标的股票”走势最相似的几只。这个策略是很多配对交易、市场中性策略的起点。

假设我们想找到与贵州茅台（600519.SH）在最近一个月走势最相似的消费股。数据准备是关键的第一步。你不能直接用原始股价，因为股价绝对值差异巨大，一只1000元的股票和一只10元的股票，DTW距离会主要被绝对值主导。我们必须先进行标准化，通常使用最小-最大归一化或Z-score标准化，将价格序列转化为主要反映涨跌形态的序列。

import numpy as np
import pandas as pd
from scipy.spatial.distance import euclidean
import yfinance as yf # 需要安装 yfinance

def normalize_series(series):
    """使用Z-score标准化"""
    return (series - series.mean()) / series.std()

# 1. 获取数据
ticker_target = '600519.SS' # 贵州茅台
tickers_pool = ['000858.SZ', '600887.SS', '000568.SZ', '603288.SS'] # 五粮液、伊利、泸州老窖、海天味业
start_date = '2024-01-01'
end_date = '2024-03-20'

# 下载标的股票数据
data_target = yf.download(ticker_target, start=start_date, end=end_date)
prices_target = data_target['Close'].values

# 2. 预处理：我们更关心价格序列的形态，所以使用收益率序列或标准化后的价格序列
# 这里使用对数收益率，更能反映相对变化，且通常更平稳
returns_target = np.diff(np.log(prices_target)) # 计算对数收益率
# 或者使用标准化价格
# norm_target = normalize_series(prices_target)

similarity_scores = []

for ticker in tickers_pool:
    data = yf.download(ticker, start=start_date, end=end_date)
    prices = data['Close'].values
    # 确保长度一致，处理缺失日期
    min_len = min(len(prices_target), len(prices))
    prices_target_trunc = prices_target[:min_len]
    prices_trunc = prices[:min_len]

    # 使用标准化后的价格进行计算
    series_a = normalize_series(prices_target_trunc)
    series_b = normalize_series(prices_trunc)

    # 3. 计算DTW距离
    dtw_dist, _ = dtw_basic(series_a, series_b) # 使用前面定义的基础DTW函数
    similarity_scores.append((ticker, dtw_dist))

# 按DTW距离从小到大排序，距离越小越相似
similarity_scores.sort(key=lambda x: x[1])
print("与贵州茅台走势相似度排名（DTW距离越小越相似）：")
for ticker, dist in similarity_scores:
    print(f"{ticker}: {dist:.4f}")

跑完这段代码，你可能会发现，和茅台走势最相似的未必是另一只白酒股，可能是调味品或乳业龙头。这很有意思，它揭示了跨细分行业的资金轮动或消费板块的整体情绪。这就是DTW发现的“形态相似性”，它超越了简单的行业分类。

但这里有个实战坑：基础DTW的计算复杂度是O(n²)。如果股票池有几百只，每只股票过去一年的日线数据（约250个点），计算一对的DTW就需要约6万次运算。几百只股票两两比较，计算量会非常大。所以，在真实的生产环境中，我们不会直接用基础DTW进行全量扫描。

4. 效率优化：让DTW在金融场景下“跑起来”

直接使用基础DTW处理金融数据，尤其是高频数据或大规模股票池，速度慢得让人无法接受。我们必须优化。好在DTW的优化策略非常成熟，主要围绕“减少搜索空间”和“近似计算”两个思路。

4.1 约束DTW：给“按摩师”划定工作区域

最直接的想法是，我们真的需要按摩师在整张“按摩床”（距离矩阵）上寻找路径吗？对于两只股票的日线走势，它们的对齐路径大概率不会偏离对角线太远。因为市场信息传导再慢，通常也不会延迟好几周。基于这个先验知识，我们可以引入Sakoe-Chiba Band（带约束）。

这个约束规定，对齐路径不能偏离对角线超过一个固定宽度 r。也就是说，序列X的第 i 个点，只能和序列Y中索引在 [i-r, i+r] 范围内的点对齐。这相当于把搜索范围从一个 m*n 的矩形，缩小为一条围绕对角线的带状区域。

def dtw_sakoe_chiba(x, y, window_ratio=0.1):
    m, n = len(x), len(y)
    window_size = int(max(m, n) * window_ratio) # 窗口大小为序列长度的10%
    dp = np.full((m, n), np.inf)
    dp[0, 0] = abs(x[0] - y[0])

    for i in range(m):
        # 关键：只计算窗口内的j
        lower_bound = max(0, i - window_size)
        upper_bound = min(n, i + window_size + 1)
        for j in range(lower_bound, upper_bound):
            if i == 0 and j == 0:
                continue
            cost = abs(x[i] - y[j])
            # 只能从也在窗口内的前驱点过来
            candidates = []
            if i > 0 and abs((i-1) - j) <= window_size:
                candidates.append(dp[i-1, j])
            if j > 0 and abs(i - (j-1)) <= window_size:
                candidates.append(dp[i, j-1])
            if i > 0 and j > 0:
                candidates.append(dp[i-1, j-1])
            dp[i, j] = cost + min(candidates)
    return dp[m-1, n-1], dp

通过设置 window_ratio（比如0.1或0.2），我们可以将计算复杂度从O(n²)降低到O(r*n)，其中r远小于n。在我的实测中，对于长度为250的序列，带约束DTW的速度可以提升5-10倍，而对最终距离计算结果的影响通常小于5%，在大多数金融分析场景下完全可以接受。

4.2 FastDTW：多分辨率“由粗到精”的搜索

另一种更精巧的思路是FastDTW。它模仿了人类看图的过程：先远观大概轮廓，再近看细节。算法首先将两个序列粗化（比如每隔两个点取一个平均值），在低分辨率上计算DTW路径。因为数据点变少了，计算飞快。然后，它将这条低分辨率的路径映射回原始分辨率，并在其周围一个狭窄的邻域内进行细化和修正。

# 这是一个简化的FastDTW思想演示，并非完整实现
def fastdtw_idea(x, y, radius=1):
    # 1. 粗化：将序列长度减半
    if len(x) > 50 or len(y) > 50: # 设置一个递归停止条件
        x_coarse = (x[::2] + x[1::2]) / 2 # 简单平均下采样
        y_coarse = (y[::2] + y[1::2]) / 2
        # 递归地在粗粒度上计算路径
        path_low_res = fastdtw_idea(x_coarse, y_coarse, radius)
        # 2. 投影：将低分辨率路径的坐标乘以2，得到原始尺度下的粗略路径
        path_initial = [(i*2, j*2) for (i, j) in path_low_res]
    else:
        # 序列足够短，直接计算基础DTW
        return dtw_basic(x, y)[1] # 返回路径

    # 3. 细化：在粗略路径的周围（radius范围内）进行局部DTW计算，优化路径
    # ... (此处省略局部优化的复杂实现)
    return refined_path

FastDTW能将平均复杂度降到O(n)，在处理长序列时优势极其明显。对于分钟级甚至Tick级数据，FastDTW几乎是唯一可行的选择。不过，它是一种近似算法，会损失少量精度来换取巨大的速度提升。在金融预测中，我们需要权衡：是追求极致的计算速度以进行实时监控，还是追求更高的精度用于离线策略回测。

4.3 下界函数：快速排除“不相似”的候选者

当我们有海量股票需要与目标股票比较时，还有一个杀手锏：下界函数。它的思想是，先用一个计算极快但较宽松的方法，估算出DTW距离的一个下限（LB）。如果这个下限已经大于我们设定的相似度阈值，那么这对序列的真实DTW距离肯定更大，可以直接淘汰，无需进行昂贵的精确DTW计算。

最常用的下界是 LB_Keogh。它为序列Y构建一个上下包络线（Envelope），序列X的点如果落在这个包络内，距离计为0，否则计为到最近包络线的距离。这个下界的计算复杂度只有O(n)，而且非常紧致（即估算的下限很接近真实DTW值）。

def lb_keogh(x, y, r=5):
    """计算序列x和序列y的LB_Keogh下界，r为包络窗口半径"""
    m = len(y)
    # 为y计算上下包络
    u = np.array([max(y[max(0, i-r):min(m, i+r+1)]) for i in range(m)])
    l = np.array([min(y[max(0, i-r):min(m, i+r+1)]) for i in range(m)])

    lb_sum = 0
    for i in range(len(x)):
        if x[i] > u[i]:
            lb_sum += (x[i] - u[i]) ** 2
        elif x[i] < l[i]:
            lb_sum += (l[i] - x[i]) ** 2
        # 落在包络内则贡献为0
    return np.sqrt(lb_sum)

# 使用示例：在股票池中快速筛选
target_series = normalize_series(prices_target_trunc)
candidate_pool = [...] # 标准化后的其他股票序列
threshold = 10.0 # 相似度阈值

potential_matches = []
for candidate in candidate_pool:
    lb = lb_keogh(target_series, candidate, r=5)
    if lb < threshold: # 只有下界小于阈值，才值得做精确DTW计算
        dtw_dist, _ = dtw_sakoe_chiba(target_series, candidate)
        if dtw_dist < threshold:
            potential_matches.append((candidate, dtw_dist))

在实际的大规模相似性搜索中，先用LB_Keogh过滤掉90%以上的不相关序列，再对剩下的少数候选进行精确DTW计算，可以带来上百倍的效率提升。

5. 精度提升：针对金融数据特点的DTW调优策略

优化了速度，我们还要追求精度。金融时间序列有其独特之处：高噪声、存在开盘收盘缺口、有特定的周期性和趋势。直接套用标准DTW可能效果不佳，需要针对性调优。

5.1 距离度量的选择：不只是欧几里得

标准DTW使用点对点的欧几里得距离。但在金融中，我们可能更关心变化率而非绝对值。因此，我们可以改用一阶差分序列（收益率）的DTW，或者直接使用能捕捉形状特征的导数DTW。

def derivative_dtw(x, y):
    """使用一阶导数（近似于序列的斜率）作为局部距离"""
    # 计算序列的近似导数
    x_deriv = np.diff(x)
    y_deriv = np.diff(y)
    # 对导数序列进行DTW计算
    return dtw_sakoe_chiba(x_deriv, y_deriv)

导数DTW能更好地匹配走势的“拐点”和“趋势”，对价格的绝对水平不敏感，更符合技术分析的理念。我曾在对比同一板块个股的突破形态时，导数DTW的效果远好于原始价格DTW。

5.2 加权DTW：给关键区域更高权重

金融序列中，某些区域可能更重要。例如，放量突破点、重要支撑阻力位、财报发布日附近的走势。我们可以给这些关键点赋予更高的权重，让DTW在匹配时更关注这些区域。

实现上，可以在计算局部距离 d(i, j) 时，乘以一个权重系数 w(i, j)。这个权重可以基于成交量（放量区域权重高）、波动率（高波动区域权重高）或自定义的关键事件窗口来设定。

def weighted_dtw(x, y, weight_matrix):
    """带权重的DTW，weight_matrix形状与距离矩阵相同"""
    m, n = len(x), len(y)
    dp = np.full((m, n), np.inf)
    dp[0, 0] = abs(x[0] - y[0]) * weight_matrix[0, 0]
    # ... 动态规划递推时，每一步的成本都乘以对应的权重
    for i in range(1, m):
        for j in range(1, n):
            cost = abs(x[i] - y[j]) * weight_matrix[i, j]
            dp[i, j] = cost + min(dp[i-1, j], dp[i, j-1], dp[i-1, j-1])
    return dp[m-1, n-1]

这需要你根据具体的分析目标来设计权重矩阵，是一种更高级、更定制化的用法。

5.3 多尺度与多特征DTW

股价序列包含多种信息：价格、成交量、资金流等。我们可以构建多维时间序列，然后使用多维DTW进行匹配。这样，DTW在比较形态时，会同时考虑价、量、资金等多个维度，找到的相似模式更稳健。

另一种思路是进行多尺度分析。先在小时间尺度（如5分钟线）上捕捉日内模式，再在日线尺度上捕捉趋势模式，最后将不同尺度的DTW距离进行加权融合。这能同时兼顾短期波动和长期趋势，避免单一尺度的片面性。

6. 实战进阶：构建基于DTW的简单预测模型

最后，我们聊点更刺激的：如何用DTW做简单的走势预测？注意，这不是一个能保证盈利的“圣杯”策略，而是一个展示DTW应用潜力的思路框架。

其核心思想是 “K近邻模式匹配”：

1. 构建历史模式库：将标的股票的历史价格序列，切割成一个个固定长度的滑动窗口（例如，60个交易日的窗口），每个窗口代表一个“历史模式”。
2. 寻找最近邻：将当前最新的一个窗口（例如最近60天）作为查询序列，用DTW在历史模式库中寻找K个最相似的过去模式（K-Nearest Neighbors）。
3. 推断未来走势：查看这K个最相似的历史模式，在它们各自出现之后，后续N天（例如未来20天）的股价平均表现如何。将这个平均表现作为对未来的一个概率性预测。

def dtw_knn_forecast(price_series, window_len=60, forecast_horizon=20, k=5):
    """
    基于DTW的K近邻预测
    price_series: 归一化后的价格序列
    window_len: 模式窗口长度
    forecast_horizon: 预测未来多少天
    k: 最近邻数量
    """
    n = len(price_series)
    historical_patterns = []
    future_returns = []

    # 1. 构建历史模式库及对应的未来收益
    for i in range(window_len, n - forecast_horizon):
        pattern = price_series[i-window_len:i]
        future_return = (price_series[i+forecast_horizon-1] / price_series[i]) - 1  # 未来N日的简单收益率
        historical_patterns.append(pattern)
        future_returns.append(future_return)

    # 2. 获取当前窗口（最近window_len天）
    current_pattern = price_series[-window_len:]

    # 3. 计算当前窗口与所有历史模式的DTW距离
    distances = []
    for idx, hist_pattern in enumerate(historical_patterns):
        dist, _ = dtw_sakoe_chiba(current_pattern, hist_pattern, window_ratio=0.1)
        distances.append((dist, idx))

    # 4. 找到K个最近邻
    distances.sort(key=lambda x: x[0])
    k_nearest_indices = [idx for _, idx in distances[:k]]

    # 5. 聚合最近邻的未来收益作为预测
    predicted_future_returns = [future_returns[idx] for idx in k_nearest_indices]
    forecast = np.mean(predicted_future_returns)  # 简单平均

    # 可以返回预测值，以及最近邻的索引用于分析
    return forecast, k_nearest_indices

# 使用示例
norm_prices = normalize_series(prices_target) # 使用标准化后的价格
pred_return, neighbor_indices = dtw_knn_forecast(norm_prices, window_len=60, forecast_horizon=10, k=3)
print(f"基于DTW-KNN模型，预测未来10个交易日的平均收益率为: {pred_return:.4f}")
print(f"参考的历史相似模式发生在索引: {neighbor_indices}")

这个模型非常直观，本质上是一种“以史为鉴”。它的预测能力严重依赖于历史是否真的会重演，以及你选择的特征（是原始价格、收益率还是其他指标）。在实际使用中，你需要：

• 谨慎评估：必须在足够长的历史数据上进行严格的回测，计算预测的准确率、夏普比率等指标。
• 结合其他因子：DTW相似度可以作为一个特征，与其他技术指标、基本面因子一起，输入到更复杂的机器学习模型（如梯度提升树）中。
• 注意过拟合：避免在短时间窗口内过度优化参数（K值、窗口长度、DTW约束半径等）。

我在早期尝试这个思路时，发现它在趋势延续性强的市场环境中（如单边牛市或熊市）表现尚可，但在震荡市中信号混乱。后来我将DTW相似度与市场波动率、成交量放大等确认信号结合使用，过滤掉了很多无效的“相似模式”，才提高了策略的稳定性。这再次印证了，在金融领域，没有一劳永逸的算法，只有不断结合领域知识进行迭代和优化的过程。