Поиск ценовых паттернов с помощью Dynamic Time Warping (DTW) в Pandas

Суть: Алгоритм Dynamic Time Warping (DTW) позволяет находить схожие по форме ценовые паттерны в исторических данных, даже если они растянуты или сжаты во времени. В отличие от евклидова расстояния, DTW сопоставляет точки нелинейно, что делает его идеальным инструментом для технического анализа и количественного трейдинга.

Содержание

Исходный код

Классическое евклидово расстояние требует строго одинаковой длины рядов и чувствительно к сдвигам во времени. Алгоритм DTW решает эту проблему, строя оптимальный путь соответствия между двумя временными рядами. Перед расчетом DTW критически важно провести Z-score нормализацию данных, чтобы сравнивать именно форму движения цены, а не ее абсолютные значения.

Для эффективной работы с большими объемами исторических данных, например, минутными барами, рекомендуется предварительно оптимизировать их хранение. В этом поможет Конвертация HDF5 в Parquet с партиционированием по датам в Pandas.

import numpy as np
import pandas as pd
from fastdtw import fastdtw
from scipy.spatial.distance import euclidean

class DTWPatternFinder:
    def __init__(self, target_pattern: pd.Series):
        """
        Инициализация класса для поиска паттернов.
        target_pattern: pd.Series - искомый паттерн (шаблон)
        """
        if target_pattern.std() == 0:
            raise ValueError("Стандартное отклонение паттерна не должно быть равно нулю.")
        
        # Z-score нормализация целевого паттерна
        self.target = (target_pattern - target_pattern.mean()) / target_pattern.std()
        self.target_len = len(target_pattern)

    def find_matches(self, data: pd.Series, step: int = 1) -> pd.DataFrame:
        """
        Поиск похожих паттернов в историческом ряду методом скользящего окна.
        data: pd.Series - исторические цены
        step: int - шаг смещения окна
        """
        results = []
        
        # Проходим скользящим окном по историческим данным
        for i in range(0, len(data) - self.target_len + 1, step):
            window = data.iloc[i : i + self.target_len]
            
            if window.std() == 0:
                continue
                
            # Z-score нормализация текущего окна
            window_norm = (window - window.mean()) / window.std()
            
            # Вычисление DTW расстояния
            distance, _ = fastdtw(self.target.values, window_norm.values, dist=euclidean)
            
            results.append({
                'start_idx': window.index[0],
                'end_idx': window.index[-1],
                'distance': distance
            })
            
        df_res = pd.DataFrame(results)
        if not df_res.empty:
            df_res = df_res.sort_values(by='distance').reset_index(drop=True)
        return df_res

# Пример использования
if __name__ == '__main__':
    # Генерируем синтетический ценовой ряд (случайное блуждание)
    np.random.seed(42)
    steps = np.random.normal(0, 1, 500)
    prices = pd.Series(np.cumsum(steps) + 100, index=pd.date_range('2023-01-01', periods=500, freq='H'))

    # Вырезаем шаблон (например, резкое падение и восстановление - V-образное дно)
    target_pattern = prices.iloc[100:130]

    # Инициализируем поиск
    finder = DTWPatternFinder(target_pattern)
    
    # Ищем совпадения в остальной части ряда (исключая сам шаблон)
    search_area = pd.concat([prices.iloc[:100], prices.iloc[130:]])
    matches = finder.find_matches(search_area, step=2)

    print("Топ-5 наиболее похожих паттернов по метрике DTW:")
    print(matches.head(5))

Разбор параметров

target_pattern: Объект pd.Series, представляющий собой эталонный паттерн, форму которого мы хотим найти в исторических данных.
data: Исторический временной ряд pd.Series (например, цены закрытия Close), в котором производится поиск совпадений.
step: Шаг скользящего окна. Увеличение step ускоряет расчеты, но снижает точность локализации границ паттерна.
fastdtw: Оптимизированная библиотека для расчета DTW с временной сложностью O(N) вместо классической O(N²), что критично для длинных выборок.
euclidean: Метрика расстояния, используемая внутри DTW для сравнения отдельных точек двух временных рядов.

Как запустить

Перед запуском скрипта необходимо установить библиотеки pandas, numpy, scipy и fastdtw. Сделать это можно через стандартный менеджер пакетов:

pip install pandas numpy scipy fastdtw

Перед тем как запускать поиск паттернов на реальном рынке, полезно оценить текущую рыночную структуру (наличие тренда или флэта). В этом вам поможет Расчет показателя Херста (Hurst Exponent) для BTC/USDT на Python. Если рынок находится в состоянии флэта, паттерны разворота (двойное дно, голова и плечи) будут отрабатывать точнее.

После того как вы найдете исторические аналоги текущего движения цены, вы можете построить на их основе прогностическую модель или торговую систему. Оценить эффективность полученной стратегии и рассчитать ключевые риск-метрики поможет Расчет Information Ratio, Alpha и Beta к Bitcoin на Python.