Бэктестинг Pairs Trading на Python: Тест Йохансена (Johansen Test)

Суть: Реализация алгоритма парного трейдинга на основе многомерного теста коинтеграции Йохансена (coint_johansen). Скрипт рассчитывает стационарный спред для двух активов, генерирует торговые сигналы по Z-score и проводит векторный бэктест доходности.

Содержание

Исходный код

В отличие от простых методов, таких как Симуляция Монте-Карло для крипто-портфеля в Python scipy, которые моделируют случайные блуждания портфеля, парный трейдинг (Pairs Trading) опирается на строгую математическую взаимосвязь активов. Тест Йохансена позволяет определить наличие коинтеграции между несколькими временными рядами и найти точные коэффициенты для построения стационарного спреда.

Для фильтрации ложных сигналов на неликвидных рынках или во время аномальных всплесков объемов можно дополнительно использовать алгоритмы машинного обучения, например, Выявление Аномалий в Торговых Объемах: Isolation Forest в Python.

Ниже представлен готовый к запуску скрипт бэктестинга на Python, использующий библиотеку statsmodels для проведения теста Йохансена и симуляции торговой стратегии.

import numpy as np
import pandas as pd
from statsmodels.tsa.vector_ar.vecm import coint_johansen

def generate_synthetic_cointegrated_data(seed=42, n=1000):
    np.random.seed(seed)
    # Общий случайный тренд (случайное блуждание)
    common_trend = np.cumsum(np.random.normal(0, 1, n)) + 100
    # Актив A
    asset_a = common_trend + np.random.normal(0, 1, n)
    # Актив B коинтегрирован с активом A
    asset_b = 1.2 * common_trend + np.random.normal(0, 1.5, n) + 10
    
    dates = pd.date_range(start='2022-01-01', periods=n, freq='D')
    return pd.DataFrame({'Asset_A': asset_a, 'Asset_B': asset_b}, index=dates)

def run_johansen_test(df, det_order=0, k_ar_diff=1):
    # Тест Йохансена ожидает на вход двумерный массив данных
    res = coint_johansen(df, det_order=det_order, k_ar_diff=k_ar_diff)
    # Извлекаем собственный вектор, соответствующий наибольшему собственному значению
    # Это дает нам коэффициенты коинтеграции для построения спреда
    evec = res.evec[:, 0]
    # Нормализуем вектор по первому активу
    beta = -evec[1] / evec[0]
    return beta, res

def backtest_pairs_trading(df, entry_threshold=1.5, exit_threshold=0.0):
    # Оценка коэффициента коинтеграции на исторических данных
    beta, johansen_res = run_johansen_test(df[['Asset_A', 'Asset_B']])
    
    # Расчет спреда: Spread = Asset_A - beta * Asset_B
    df['Spread'] = df['Asset_A'] - beta * df['Asset_B']
    
    # Расчет скользящего Z-score для избежания look-ahead bias
    window = 30
    df['Spread_Mean'] = df['Spread'].rolling(window).mean()
    df['Spread_Std'] = df['Spread'].rolling(window).std()
    df['Z_Score'] = (df['Spread'] - df['Spread_Mean']) / df['Spread_Std']
    
    # Генерация сигналов
    df['Signal'] = 0
    # Покупка спреда (Long Asset_A, Short Asset_B)
    df.loc[df['Z_Score'] < -entry_threshold, 'Signal'] = 1
    # Продажа спреда (Short Asset_A, Long Asset_B)
    df.loc[df['Z_Score'] > entry_threshold, 'Signal'] = -1
    
    # Логика удержания и выхода из позиций
    position = 0
    positions = []
    for i in range(len(df)):
        z = df['Z_Score'].iloc[i]
        sig = df['Signal'].iloc[i]
        
        if pd.isna(z):
            positions.append(0)
            continue
            
        if position == 0:
            if sig != 0:
                position = sig
        else:
            # Выход при пересечении целевого уровня (exit_threshold)
            if position == 1 and z >= exit_threshold:
                position = 0
            elif position == -1 and z <= -exit_threshold:
                position = 0
        positions.append(position)
        
    df['Position'] = positions
    
    # Расчет доходности активов
    df['Ret_A'] = df['Asset_A'].pct_change()
    df['Ret_B'] = df['Asset_B'].pct_change()
    
    # Доходность спреда с учетом весов (beta)
    df['Spread_Return'] = df['Ret_A'] - beta * df['Ret_B']
    df['Strategy_Return'] = df['Position'].shift(1) * df['Spread_Return']
    
    # Кумулятивная доходность
    df['Cum_Strategy_Return'] = (1 + df['Strategy_Return'].fillna(0)).cumprod() - 1
    return df, beta

if __name__ == '__main__':
    # Генерируем данные
    data = generate_synthetic_cointegrated_data()
    
    # Запускаем бэктест
    results, beta_coeff = backtest_pairs_trading(data)
    
    print(f'Расчитанный коэффициент коинтеграции (Beta): {beta_coeff:.4f}')
    print(f'Финальная доходность стратегии: {results["Cum_Strategy_Return"].iloc[-1] * 100:.2f}%')

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

det_order: Определяет наличие детерминированных трендов в данных. Значение -1 означает отсутствие константы, 0 — наличие константы (ограниченный дрейф), 1 — наличие линейного тренда.
k_ar_diff: Количество лагов авторегрессии в первых разностях. Этот параметр критически важен для устранения автокорреляции в остатках модели VAR (Vector Autoregression).
entry_threshold: Порог входа в позицию, выраженный в количестве стандартных отклонений скользящего Z-score. Обычно выбирается в диапазоне от 1.5 до 2.0.
exit_threshold: Порог выхода из позиции. Значение 0.0 означает закрытие сделки при возврате спреда к своему среднему значению.

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

Для запуска скрипта вам понадобятся библиотеки numpy, pandas и statsmodels. Установите их с помощью пакетного менеджера:

pip install numpy pandas statsmodels

Сохраните приведенный выше код в файл pairs_backtest.py и запустите его в терминале. Скрипт автоматически сгенерирует коинтегрированные временные ряды, рассчитает вектор Йохансена, построит торговые сигналы и выведет итоговую доходность бэктеста.

Если вы хотите расширить свой арсенал системного трейдинга за пределы линейных спотовых инструментов, рекомендуем изучить Бэктест опционных стратегий Iron Condor на Python с Implied Volatility.