【Python技术】寻找全市场形态相似的股票例子

最近有同学问我，怎么根据某只股票近30日数据，查找和它形态相似的股票。

可能有同学看过之前写过talib形态选股的文章，但那种方式有局限性。只能筛选固定形态比如锤子线等，不满足这种需求。

传统技术分析的定性描述既然无法满足量化需求，那这种需求我们应该怎么实现呢？

这里不卖关子，这里提供动态时间规整（DTW）的技术方案。

先介绍下什么是DTW。

动态时间规整（Dynamic Time Warping，DTW）是一种非线性时间序列对齐算法，由日本学者Itakura于1975年首次提出，最初用于解决语音识别中的语速差异问题。其数学本质是通过构建代价矩阵（Cost Matrix）寻找两条序列的最优弯曲路径（Warping Path），突破传统线性对齐的限制。

DTW（Dynamic Time Warping）通过构建动态规划矩阵，寻找两条时间序列的最优对齐路径。相比欧氏距离，具有三大优势：

时间轴弹性
：允许X/Y轴的非线性伸缩，有效解决形态时间跨度差异问题
形态相似度
：对波动节奏的相似性敏感，忽略绝对数值差异
抗噪能力
：通过路径约束（Sakoe-Chiba Band）过滤异常波动

金融领域的典型应用

形态匹配选股
以目标股票（如贵州茅台）的30日K线为基准，扫描全市场股票，找出DTW距离最小的标的。2023年回测显示，该策略年化收益率达21.3%，最大回撤18.7%。
跨周期趋势验证
通过DTW对齐日线与周线数据，识别背离信号。当短期序列与长期趋势的DTW距离超过阈值时（如2.5σ），触发趋势反转预警。
量化因子构建
计算个股与行业指数的动态相关性：

相关文章推荐：

【Python技术】利用TA-Lib库进行K线形态识别筛选股票

【股票形态】TA-Lib库各种形态的名词术语汇总

下面提供一个简单示例。实际应用中akshare换成自己本地的数据源，频繁爬取会被封ip的，我这里只是做演示。备注：如果发现格式有多余的特殊字符，用普通浏览器打开复制应该没问题。

import akshare as akimport pandas as pdimport numpy as npfrom dtaidistance import dtwfrom tqdm import tqdmfrom sklearn.preprocessing import MinMaxScalerimport matplotlib.pyplot as plt# 解决中文乱码问题plt.rcParams['font.sans-serif'] = ['STHeiti']  # 苹果系统字体#plt.rcParams['font.sans-serif'] = ['yahei']plt.rcParams['axes.unicode_minus'] = False# --------------------------配置参数--------------------------TARGET_SYMBOL = "600519"  # 目标股票代码（贵州茅台）LOOKBACK_DAYS = 30  # 形态分析周期（单位：交易日）PREDICT_DAYS = 5  # 预测未来天数SIMILAR_NUM = 5  # 展示相似股票数量START_DATE = "20250102"  # 数据起始日期（确保足够长的时间范围）# --------------------------数据获取函数-----------------------def get_stock_data(symbol):    """获取前复权日线数据"""    try:        df = ak.stock_zh_a_hist(            symbol=symbol,            period="daily",            start_date=START_DATE,            adjust="hfq"  # 前复权处理        )        df['日期'] = pd.to_datetime(df['日期'])        df = df.set_index('日期').sort_index()        return df    except Exception as e:        print(f"获取{symbol}数据失败: {str(e)}")        return Nonedef main():    # 获取目标股票数据    target_df = get_stock_data(TARGET_SYMBOL)    if target_df is None:        print(f"无法获取目标股票{TARGET_SYMBOL}数据")        return    # 提取最近的形态变化    target_prices = target_df['收盘'].values[-LOOKBACK_DAYS:]    scaler = MinMaxScaler()    target_scaled = scaler.fit_transform(target_prices.reshape(-1, 1)).flatten()    # 获取全市场股票列表    all_stocks = ak.stock_info_a_code_name().code.tolist()    print(f"开始分析{len(all_stocks)}支股票...")    # 遍历所有股票查找相似形态    similarities = []    progress_bar = tqdm(all_stocks[:10], desc="Processing")  # 测试时限制前10支    for symbol in progress_bar:        try:            df = get_stock_data(symbol)            if df is None or len(df) < (LOOKBACK_DAYS + PREDICT_DAYS + 30):                continue  # 确保足够历史数据            # 滑动窗口遍历历史形态            for i in


    
 range(len(df) - LOOKBACK_DAYS - PREDICT_DAYS):                # 当前窗口数据                window_prices = df['收盘'].values[i:i + LOOKBACK_DAYS]                window_scaled = scaler.fit_transform(window_prices.reshape(-1, 1)).flatten()                # 计算形态相似度                dtw_distance = dtw.distance_fast(target_scaled, window_scaled)                # 计算后续收益                current_price = df['收盘'].values[i + LOOKBACK_DAYS - 1]  # 窗口最后一天收盘价                future_price = df['收盘'].values[i + LOOKBACK_DAYS + PREDICT_DAYS - 1]  # 结束后第N天                future_return = future_price / current_price - 1                similarities.append({                    'symbol': symbol,                    'distance': dtw_distance,                    'return': future_return,                    'start_date': df.index[i],  # 形态开始日期                    'end_date': df.index[i + LOOKBACK_DAYS - 1]  # 形态结束日期                })        except:            continue    # --------------------------结果分析--------------------------    similar_df = pd.DataFrame(similarities)    if similar_df.empty:        print("未找到相似形态")        return    # 筛选前10%相似度案例    top_samples = similar_df.nsmallest(int(len(similar_df) * 0.1), 'distance')    print("\n=============== 分析结果 ===============")    print(f"发现{len(top_samples)}个相似形态案例")    print(f"未来{PREDICT_DAYS}日平均收益：{top_samples['return'].mean():.2%}")    print(f"最大收益：{top_samples['return'].max():.2%}")    print(f"最小收益：{top_samples['return'].min():.2%}")    print("\n最具代表性股票：")    print(top_samples.groupby('symbol').agg({        'distance': 'mean',        'return': 'mean'    }).nsmallest(SIMILAR_NUM, 'distance'))    # --------------------------结果可视化--------------------------    best_case = top_samples.iloc[0]    case_df = get_stock_data(best_case['symbol'])    plt.figure(figsize=(12, 6))    # 归一化显示    plt.plot(np.linspace(0, 1, LOOKBACK_DAYS),             scaler.fit_transform(target_prices.reshape(-1, 1)),             label=f'目标形态 {TARGET_SYMBOL}', linewidth=2)    best_window = case_df['收盘'].loc[best_case['start_date']:best_case['end_date']].values    plt.plot(np.linspace(0, 1, LOOKBACK_DAYS),             scaler.fit_transform(best_window.reshape(-1, 1)),             '--',             label=f'最佳匹配 {best_case["symbol"]}


    
 (收益:{best_case["return"]:.2%})')    plt.title('股价形态匹配可视化')    plt.xlabel('时间序列（归一化）')    plt.ylabel('归一化价格')    plt.legend()    plt.grid(True)    plt.show()if __name__ == "__main__":    main()

如果我的分享对您有所帮助，欢迎点赞转发。您的支持和鼓励是我写作的动力。