Дистанционный подход в парном трейдинге: реализация и анализ на Rust

Дистанционный подход в парном трейдинге приобрёл большую популярность благодаря своей элегантной простоте и эффективности. Эта техника определяет пары активов с помощью статистических мер и осуществляет сделки на основе расхождения и схождения их ценовых отношений. В этой статье представлен комплексный анализ как базовых, так и продвинутых методик дистанционного подхода с практическими реализациями на Rust, предназначенными для высокочастотных трейдеров, алготрейдеров, математиков и программистов, ищущих надёжные решения.

Теоретические основы дистанционного подхода

Дистанционный подход формирует основу для парного трейдинга на основе нормализованных движений цен между активами. В его основе лежит использование евклидовой квадратичной дистанции для выявления активов, которые исторически движутся вместе, и генерации торговых сигналов, когда их нормализованное ценовое расхождение превышает статистически значимые пороги[2].

Этот подход состоит из двух основных этапов:

Формирование пар — выявление статистически связанных пар активов
Генерация торговых сигналов — создание правил входа и выхода на основе расхождения

Математическая основа

Базовая реализация использует евклидово расстояние между нормализованными временными рядами цен. Для двух активов с нормализованными временными рядами X и Y вычисляется:

fn euclidean_squared_distance(x: &[f64], y: &[f64]) -> f64 {
    assert_eq!(x.len(), y.len(), "Time series must have equal length");
    
    x.iter()
        .zip(y.iter())
        .map(|(xi, yi)| (xi - yi).powi(2))
        .sum()
}

Эта метрика позволяет выявлять активы, которые исторически движутся вместе, что создаёт основу для статистического арбитража[2].

Базовая реализация дистанционного подхода

Нормализация данных

Перед вычислением расстояний необходимо нормализовать ценовые данные для сопоставимости масштабов. Обычно применяется min-max нормализация:

fn min_max_normalize(prices: &[f64]) -> Vec<f64> {
    if prices.is_empty() {
        return Vec::new();
    }
    
    let min_price = prices.iter().fold(f64::INFINITY, |a, &b| a.min(b));
    let max_price = prices.iter().fold(f64::NEG_INFINITY, |a, &b| a.max(b));
    let range = max_price - min_price;
    
    if range.abs() < f64::EPSILON {
        return vec![0.5; prices.len()];
    }
    
    prices.iter()
        .map(|&price| (price - min_price) / range)
        .collect()
}

Поиск ближайших пар

Потенциальные пары определяются путём вычисления евклидова расстояния между всеми комбинациями активов и выбора тех, у которых расстояние наименьшее:

#[derive(Debug, Clone)]
struct StockPair {
    stock1_idx: usize,
    stock2_idx: usize,
    distance: f64,
}

impl PartialEq for StockPair {
    fn eq(&self, other: &Self) -> bool {
        self.distance.eq(&other.distance)
    }
}

impl Eq for StockPair {}

impl PartialOrd for StockPair {
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
        self.distance.partial_cmp(&other.distance)
    }
}

impl Ord for StockPair {
    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
        self.partial_cmp(other).unwrap_or(std::cmp::Ordering::Equal)
    }
}

fn find_closest_pairs(normalized_prices: &[Vec<f64>], top_n: usize) -> Vec<StockPair> {
    let stock_count = normalized_prices.len();
    let mut pairs = BinaryHeap::new();
    
    for i in 0..stock_count {
        for j in (i+1)..stock_count {
            let distance = euclidean_squared_distance(&normalized_prices[i], &normalized_prices[j]);
            
            pairs.push(Reverse(StockPair {
                stock1_idx: i,
                stock2_idx: j,
                distance,
            }));
            
            // Keep only top N pairs
            if pairs.len() > top_n {
                pairs.pop();
            }
        }
    }
    
    // Convert from heap to vector and reverse to get ascending order
    pairs.into_iter().map(|Reverse(pair)| pair).collect()
}

Расчёт исторической волатильности

Расчёт исторической волатильности необходим для установки корректных торговых порогов:

fn calculate_spread_volatility(normalized_price1: &[f64], normalized_price2: &[f64]) -> f64 {
    assert_eq!(normalized_price1.len(), normalized_price2.len());
    
    // Calculate price spread
    let spread: Vec<f64> = normalized_price1.iter()
        .zip(normalized_price2.iter())
        .map(|(p1, p2)| p1 - p2)
        .collect();
    
    // Calculate mean of spread
    let mean = spread.iter().sum::<f64>() / spread.len() as f64;
    
    // Calculate standard deviation
    let variance = spread.iter()
        .map(|&x| (x - mean).powi(2))
        .sum::<f64>() / spread.len() as f64;
    
    variance.sqrt()
}

Продвинутые методы отбора

Фильтрация по отраслевым группам

Ограничение отбора пар одной отраслью может повысить эффективность за счёт выбора экономически связанных активов:

fn find_industry_pairs(
    normalized_prices: &[Vec<f64>], 
    industry_codes: &[usize], 
    top_n_per_industry: usize
) -> Vec<StockPair> {
    // Group stocks by industry
    let mut industry_groups: std::collections::HashMap<usize, Vec<usize>> = std::collections::HashMap::new();
    
    for (idx, &code) in industry_codes.iter().enumerate() {
        industry_groups.entry(code).or_default().push(idx);
    }
    
    // Find closest pairs within each industry
    let mut all_pairs = Vec::new();
    
    for (_industry_code, stock_indices) in industry_groups {
        let mut industry_pairs = Vec::new();
        
        for i in 0..stock_indices.len() {
            for j in (i+1)..stock_indices.len() {
                let stock1_idx = stock_indices[i];
                let stock2_idx = stock_indices[j];
                let distance = euclidean_squared_distance(
                    &normalized_prices[stock1_idx], 
                    &normalized_prices[stock2_idx]
                );
                
                industry_pairs.push(StockPair {
                    stock1_idx,
                    stock2_idx,
                    distance,
                });
            }
        }
        
        // Sort pairs by distance
        industry_pairs.sort_by(|a, b| a.distance.partial_cmp(&b.distance).unwrap());
        
        // Take top N from each industry
        let top_pairs: Vec<StockPair> = industry_pairs.into_iter()
            .take(top_n_per_industry)
            .collect();
            
        all_pairs.extend(top_pairs);
    }
    
    all_pairs
}

Метод нулевых пересечений

Метод нулевых пересечений выявляет пары с частыми схождениями и расхождениями, что потенциально может указывать на более прибыльные торговые возможности:

fn count_zero_crossings(spread: &[f64]) -> usize {
    if spread.len() < 2 {
        return 0;
    }
    
    let mut count = 0;
    for i in 1..spread.len() {
        if (spread[i-1] < 0.0 && spread[i] >= 0.0) || 
           (spread[i-1] >= 0.0 && spread[i] < 0.0) {
            count += 1;
        }
    }
    
    count
}

fn find_zero_crossing_pairs(
    normalized_prices: &[Vec<f64>],
    top_distance_threshold: f64,
    min_crossings: usize
) -> Vec<StockPair> {
    let stock_count = normalized_prices.len();
    let mut qualifying_pairs = Vec::new();
    
    for i in 0..stock_count {
        for j in (i+1)..stock_count {
            let distance = euclidean_squared_distance(&normalized_prices[i], &normalized_prices[j]);
            
            // Only consider pairs with distance below threshold
            if distance < top_distance_threshold {
                // Calculate spread
                let spread: Vec<f64> = normalized_prices[i].iter()
                    .zip(normalized_prices[j].iter())
                    .map(|(p1, p2)| p1 - p2)
                    .collect();
                
                let crossings = count_zero_crossings(&spread);
                
                if crossings >= min_crossings {
                    qualifying_pairs.push(StockPair {
                        stock1_idx: i,
                        stock2_idx: j,
                        distance,
                    });
                }
            }
        }
    }
    
    // Sort by number of crossings (could extend StockPair to include this)
    qualifying_pairs.sort_by(|a, b| a.distance.partial_cmp(&b.distance).unwrap());
    qualifying_pairs
}

Учёт исторического стандартного отклонения

Этот метод устраняет ограничение базового подхода, отдавая приоритет парам с большей волатильностью спреда, что может повысить потенциальную прибыль:

fn find_highsd_pairs(
    normalized_prices: &[Vec<f64>],
    top_distance_count: usize,
    min_volatility: f64
) -> Vec<StockPair> {
    let stock_count = normalized_prices.len();
    let mut all_pairs = Vec::new();
    
    for i in 0..stock_count {
        for j in (i+1)..stock_count {
            let distance = euclidean_squared_distance(&normalized_prices[i], &normalized_prices[j]);
            
            // Calculate spread volatility
            let spread: Vec<f64> = normalized_prices[i].iter()
                .zip(normalized_prices[j].iter())
                .map(|(p1, p2)| p1 - p2)
                .collect();
            
            let volatility = calculate_spread_volatility(&normalized_prices[i], &normalized_prices[j]);
            
            if volatility >= min_volatility {
                all_pairs.push(StockPair {
                    stock1_idx: i,
                    stock2_idx: j,
                    distance,
                });
            }
        }
    }
    
    // Sort by distance
    all_pairs.sort_by(|a, b| a.distance.partial_cmp(&b.distance).unwrap());
    
    // Take top N pairs with highest volatility that meet distance criteria
    all_pairs.into_iter().take(top_distance_count).collect()
}

Продвинутый подход: метод корреляции Пирсона

Метод корреляции Пирсона предлагает ряд преимуществ по сравнению с базовым дистанционным подходом, фокусируясь на корреляции доходностей, а не на ценовых расстояниях[1].

Реализация на Rust

fn pearson_correlation(x: &[f64], y: &[f64]) -> f64 {
    assert_eq!(x.len(), y.len(), "Arrays must have the same length");
    
    let n = x.len() as f64;
    let sum_x: f64 = x.iter().sum();
    let sum_y: f64 = y.iter().sum();
    let sum_xx: f64 = x.iter().map(|&val| val * val).sum();
    let sum_yy: f64 = y.iter().map(|&val| val * val).sum();
    let sum_xy: f64 = x.iter().zip(y.iter()).map(|(&xi, &yi)| xi * yi).sum();
    
    let numerator = n * sum_xy - sum_x * sum_y;
    let denominator = ((n * sum_xx - sum_x * sum_x) * (n * sum_yy - sum_y * sum_y)).sqrt();
    
    if denominator.abs() < f64::EPSILON {
        return 0.0;
    }
    
    numerator / denominator
}

struct PearsonPair {
    stock_idx: usize,
    comover_indices: Vec<usize>,
    correlations: Vec<f64>,
}

fn find_pearson_pairs(returns: &[Vec<f64>], top_n_comovers: usize) -> Vec<PearsonPair> {
    let stock_count = returns.len();
    let mut all_pairs = Vec::new();
    
    for i in 0..stock_count {
        let mut correlations = Vec::with_capacity(stock_count - 1);
        
        for j in 0..stock_count {
            if i == j {
                continue;
            }
            
            let correlation = pearson_correlation(&returns[i], &returns[j]).abs();
            correlations.push((j, correlation));
        }
        
        // Sort by correlation (highest first)
        correlations.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal));
        
        // Take top N comovers
        let top_comovers: Vec<(usize, f64)> = correlations.into_iter()
            .take(top_n_comovers)
            .collect();
            
        let (comover_indices, correlation_values): (Vec<usize>, Vec<f64>) = 
            top_comovers.into_iter().unzip();
            
        all_pairs.push(PearsonPair {
            stock_idx: i,
            comover_indices,
            correlations: correlation_values,
        });
    }
    
    all_pairs
}

Формирование портфеля и расчёт беты

Подход Пирсона формирует портфели из наиболее коррелирующих активов для каждого инструмента, а затем рассчитывает коэффициенты регрессии:

fn calculate_beta(stock_returns: &[f64], portfolio_returns: &[f64]) -> f64 {
    let cov_xy = covariance(stock_returns, portfolio_returns);
    let var_x = variance(portfolio_returns);
    
    if var_x.abs() < f64::EPSILON {
        return 0.0;
    }
    
    cov_xy / var_x
}

fn covariance(x: &[f64], y: &[f64]) -> f64 {
    assert_eq!(x.len(), y.len());
    
    let n = x.len() as f64;
    let mean_x: f64 = x.iter().sum::<f64>() / n;
    let mean_y: f64 = y.iter().sum::<f64>() / n;
    
    let sum_cov: f64 = x.iter()
        .zip(y.iter())
        .map(|(&xi, &yi)| (xi - mean_x) * (yi - mean_y))
        .sum();
    
    sum_cov / n
}

fn variance(x: &[f64]) -> f64 {
    let n = x.len() as f64;
    let mean: f64 = x.iter().sum::<f64>() / n;
    
    let sum_var: f64 = x.iter()
        .map(|&xi| (xi - mean).powi(2))
        .sum();
    
    sum_var / n
}

Генерация торговых сигналов

Заключительный этап обоих подходов — генерация торговых сигналов на основе порогов расхождения:

enum TradingSignal {
    Long,
    Short,
    Neutral
}

struct TradePosition {
    stock1_idx: usize,
    stock2_idx: usize,
    signal: TradingSignal,
    entry_spread: f64,
    timestamp: usize,
}

fn generate_trading_signals(
    normalized_prices: &[Vec<f64>],
    pairs: &[StockPair], 
    threshold_multiplier: f64,
    volatilities: &[f64],
    current_time: usize
) -> Vec<TradePosition> {
    let mut positions = Vec::new();
    
    for (pair_idx, pair) in pairs.iter().enumerate() {
        let stock1_idx = pair.stock1_idx;
        let stock2_idx = pair.stock2_idx;
        
        // Calculate current spread
        let current_spread = normalized_prices[stock1_idx][current_time] - 
                             normalized_prices[stock2_idx][current_time];
                             
        let threshold = threshold_multiplier * volatilities[pair_idx];
        
        let signal = if current_spread > threshold {
            // Stock1 is overvalued relative to Stock2
            TradingSignal::Short
        } else if current_spread < -threshold {
            // Stock1 is undervalued relative to Stock2
            TradingSignal::Long
        } else {
            TradingSignal::Neutral
        };
        
        if signal != TradingSignal::Neutral {
            positions.push(TradePosition {
                stock1_idx,
                stock2_idx,
                signal,
                entry_spread: current_spread,
                timestamp: current_time,
            });
        }
    }
    
    positions
}

Оптимизация производительности

Для высокочастотных торговых систем производительность критична. SIMD (Single Instruction, Multiple Data) инструкции могут значительно ускорить вычисления дистанций:

#[cfg(target_arch = "x86_64")]
use std::arch::x86_64::*;

#[cfg(target_arch = "x86_64")]
#[inline]
unsafe fn euclidean_distance_simd(x: &[f32], y: &[f32]) -> f32 {
    assert_eq!(x.len(), y.len());
    
    let mut sum = _mm256_setzero_ps();
    let chunks = x.len() / 8;
    
    for i in 0..chunks {
        let xi = _mm256_loadu_ps(&x[i * 8]);
        let yi = _mm256_loadu_ps(&y[i * 8]);
        
        let diff = _mm256_sub_ps(xi, yi);
        let squared = _mm256_mul_ps(diff, diff);
        sum = _mm256_add_ps(sum, squared);
    }
    
    // Handle the remaining elements
    let mut result = _mm256_reduce_add_ps(sum);
    
    for i in (chunks * 8)..x.len() {
        result += (x[i] - y[i]).powi(2);
    }
    
    result.sqrt()
}

// Helper function to sum SIMD vector
#[cfg(target_arch = "x86_64")]
#[inline(always)]
unsafe fn _mm256_reduce_add_ps(v: __m256) -> f32 {
    let hilow = _mm256_extractf128_ps(v, 1);
    let low = _mm256_castps256_ps128(v);
    let sum128 = _mm_add_ps(hilow, low);
    
    let hi64 = _mm_extractf128_si128(_mm_castps_si128(sum128), 1);
    let low64 = _mm_castps_si128(sum128);
    let sum64 = _mm_add_ps(_mm_castsi128_ps(hi64), _mm_castsi128_ps(low64));
    
    _mm_cvtss_f32(_mm_hadd_ps(sum64, sum64))
}

Асинхронная обработка может дополнительно повысить пропускную способность, особенно при работе с множеством пар акций:

use tokio::task;
use futures::future::join_all;

async fn process_pairs_async(
    normalized_prices: &[Vec<f64>],
    stock_count: usize,
    chunk_size: usize
) -> Vec<StockPair> {
    let mut tasks = Vec::new();
    
    // Split work into chunks
    let chunks = (stock_count + chunk_size - 1) / chunk_size;
    
    for chunk in 0..chunks {
        let start = chunk * chunk_size;
        let end = std::cmp::min((chunk + 1) * chunk_size, stock_count);
        
        let prices_clone = normalized_prices.to_vec();
        
        let task = task::spawn(async move {
            let mut pairs = Vec::new();
            
            for i in start..end {
                for j in (i+1)..stock_count {
                    let distance = euclidean_squared_distance(&prices_clone[i], &prices_clone[j]);
                    
                    pairs.push(StockPair {
                        stock1_idx: i,
                        stock2_idx: j,
                        distance,
                    });
                }
            }
            
            pairs
        });
        
        tasks.push(task);
    }
    
    // Await all tasks and combine results
    let results = join_all(tasks).await;
    
    let mut all_pairs = Vec::new();
    for result in results {
        if let Ok(pairs) = result {
            all_pairs.extend(pairs);
        }
    }
    
    // Sort by distance
    all_pairs.sort_by(|a, b| a.distance.partial_cmp(&b.distance).unwrap_or(std::cmp::Ordering::Equal));
    all_pairs
}

Тестирование реализации стратегии

Для оценки реализации необходима соответствующая инфраструктура тестирования:

#[cfg(test)]
mod tests {
    use super::*;
    
    #[test]
    fn test_normalization() {
        let prices = vec![10.0, 15.0, 12.0, 18.0, 20.0];
        let normalized = min_max_normalize(&prices);
        
        let expected = vec![0.0, 0.5, 0.2, 0.8, 1.0];
        
        for (a, b) in normalized.iter().zip(expected.iter()) {
            assert!((a - b).abs() < 0.001);
        }
    }
    
    #[test]
    fn test_euclidean_distance() {
        let x = vec![0.1, 0.2, 0.3, 0.4, 0.5];
        let y = vec![0.15, 0.22, 0.35, 0.38, 0.53];
        
        let distance = euclidean_squared_distance(&x, &y);
        let expected = 0.0049; // Calculated manually
        
        assert!((distance - expected).abs() < 0.0001);
    }
    
    #[test]
    fn test_pearson_correlation() {
        let x = vec![1.0, 2.0, 3.0, 4.0, 5.0];
        let y = vec![5.0, 4.0, 3.0, 2.0, 1.0];
        
        let corr = pearson_correlation(&x, &y);
        let expected = -1.0; // Perfect negative correlation
        
        assert!((corr - expected).abs() < 0.0001);
    }
    
    // Integration tests would be implemented in tests/ directory
}

Для интеграционного тестирования рекомендуется следовать практике Rust и размещать тесты в отдельной директории tests в корне проекта[15][18].

Заключение

Дистанционный подход предоставляет надёжную основу для парного трейдинга, причём как базовые, так и продвинутые методики открывают ценные возможности для статистического арбитража. Базовый подход, фокусирующийся на евклидовой дистанции, отличается простотой и эффективностью, а метод корреляции Пирсона даёт дополнительную гибкость и, возможно, лучшие характеристики возврата к среднему.

Высокая производительность Rust делает его идеальным языком для реализации подобных вычислительно интенсивных стратегий, особенно с применением SIMD и параллельной обработки. Сочетание статистической строгости и эффективной реализации создаёт мощный инструментарий для алготрейдеров.

При реализации системы парного трейдинга важно учитывать:

Баланс между простотой (базовый подход) и статистической мощностью (подход Пирсона)
Необходимые вычислительные ресурсы для анализа большого числа пар
Транзакционные издержки, которые могут существенно влиять на прибыльность[3]
Необходимость постоянного мониторинга и перекалибровки пар

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

Цитирование

@software{soloviov2025distanceapproach,
  author = {Soloviov, Eugen},
  title = {Distance Approach in Pairs Trading: Implementation and Analysis with Rust},
  year = {2025},
  url = {https://marketmaker.cc/en/blog/post/distance-approach-pairs-trading},
  version = {0.1.0},
  description = {A comprehensive analysis of basic and advanced Distance Approach methodologies for pairs trading, with practical implementations in Rust tailored for high-frequency traders and algorithmic developers.}
}

Источники

Теоретические основы дистанционного подхода

Этот подход состоит из двух основных этапов:

Формирование пар — выявление статистически связанных пар активов
Генерация торговых сигналов — создание правил входа и выхода на основе расхождения

Математическая основа

fn euclidean_squared_distance(x: &[f64], y: &[f64]) -> f64 {
    assert_eq!(x.len(), y.len(), "Time series must have equal length");
    
    x.iter()
        .zip(y.iter())
        .map(|(xi, yi)| (xi - yi).powi(2))
        .sum()
}

Базовая реализация дистанционного подхода

Нормализация данных

fn min_max_normalize(prices: &[f64]) -> Vec<f64> {
    if prices.is_empty() {
        return Vec::new();
    }
    
    let min_price = prices.iter().fold(f64::INFINITY, |a, &b| a.min(b));
    let max_price = prices.iter().fold(f64::NEG_INFINITY, |a, &b| a.max(b));
    let range = max_price - min_price;
    
    if range.abs() < f64::EPSILON {
        return vec![0.5; prices.len()];
    }
    
    prices.iter()
        .map(|&price| (price - min_price) / range)
        .collect()
}

Поиск ближайших пар

#[derive(Debug, Clone)]
struct StockPair {
    stock1_idx: usize,
    stock2_idx: usize,
    distance: f64,
}

impl PartialEq for StockPair {
    fn eq(&self, other: &Self) -> bool {
        self.distance.eq(&other.distance)
    }
}

impl Eq for StockPair {}

impl PartialOrd for StockPair {
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
        self.distance.partial_cmp(&other.distance)
    }
}

impl Ord for StockPair {
    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
        self.partial_cmp(other).unwrap_or(std::cmp::Ordering::Equal)
    }
}

fn find_closest_pairs(normalized_prices: &[Vec<f64>], top_n: usize) -> Vec<StockPair> {
    let stock_count = normalized_prices.len();
    let mut pairs = BinaryHeap::new();
    
    for i in 0..stock_count {
        for j in (i+1)..stock_count {
            let distance = euclidean_squared_distance(&normalized_prices[i], &normalized_prices[j]);
            
            pairs.push(Reverse(StockPair {
                stock1_idx: i,
                stock2_idx: j,
                distance,
            }));
            
            // Keep only top N pairs
            if pairs.len() > top_n {
                pairs.pop();
            }
        }
    }
    
    // Convert from heap to vector and reverse to get ascending order
    pairs.into_iter().map(|Reverse(pair)| pair).collect()
}

Расчёт исторической волатильности

Расчёт исторической волатильности необходим для установки корректных торговых порогов:

fn calculate_spread_volatility(normalized_price1: &[f64], normalized_price2: &[f64]) -> f64 {
    assert_eq!(normalized_price1.len(), normalized_price2.len());
    
    // Calculate price spread
    let spread: Vec<f64> = normalized_price1.iter()
        .zip(normalized_price2.iter())
        .map(|(p1, p2)| p1 - p2)
        .collect();
    
    // Calculate mean of spread
    let mean = spread.iter().sum::<f64>() / spread.len() as f64;
    
    // Calculate standard deviation
    let variance = spread.iter()
        .map(|&x| (x - mean).powi(2))
        .sum::<f64>() / spread.len() as f64;
    
    variance.sqrt()
}

Продвинутые методы отбора

Фильтрация по отраслевым группам

fn find_industry_pairs(
    normalized_prices: &[Vec<f64>], 
    industry_codes: &[usize], 
    top_n_per_industry: usize
) -> Vec<StockPair> {
    // Group stocks by industry
    let mut industry_groups: std::collections::HashMap<usize, Vec<usize>> = std::collections::HashMap::new();
    
    for (idx, &code) in industry_codes.iter().enumerate() {
        industry_groups.entry(code).or_default().push(idx);
    }
    
    // Find closest pairs within each industry
    let mut all_pairs = Vec::new();
    
    for (_industry_code, stock_indices) in industry_groups {
        let mut industry_pairs = Vec::new();
        
        for i in 0..stock_indices.len() {
            for j in (i+1)..stock_indices.len() {
                let stock1_idx = stock_indices[i];
                let stock2_idx = stock_indices[j];
                let distance = euclidean_squared_distance(
                    &normalized_prices[stock1_idx], 
                    &normalized_prices[stock2_idx]
                );
                
                industry_pairs.push(StockPair {
                    stock1_idx,
                    stock2_idx,
                    distance,
                });
            }
        }
        
        // Sort pairs by distance
        industry_pairs.sort_by(|a, b| a.distance.partial_cmp(&b.distance).unwrap());
        
        // Take top N from each industry
        let top_pairs: Vec<StockPair> = industry_pairs.into_iter()
            .take(top_n_per_industry)
            .collect();
            
        all_pairs.extend(top_pairs);
    }
    
    all_pairs
}

Метод нулевых пересечений

fn count_zero_crossings(spread: &[f64]) -> usize {
    if spread.len() < 2 {
        return 0;
    }
    
    let mut count = 0;
    for i in 1..spread.len() {
        if (spread[i-1] < 0.0 && spread[i] >= 0.0) || 
           (spread[i-1] >= 0.0 && spread[i] < 0.0) {
            count += 1;
        }
    }
    
    count
}

fn find_zero_crossing_pairs(
    normalized_prices: &[Vec<f64>],
    top_distance_threshold: f64,
    min_crossings: usize
) -> Vec<StockPair> {
    let stock_count = normalized_prices.len();
    let mut qualifying_pairs = Vec::new();
    
    for i in 0..stock_count {
        for j in (i+1)..stock_count {
            let distance = euclidean_squared_distance(&normalized_prices[i], &normalized_prices[j]);
            
            // Only consider pairs with distance below threshold
            if distance < top_distance_threshold {
                // Calculate spread
                let spread: Vec<f64> = normalized_prices[i].iter()
                    .zip(normalized_prices[j].iter())
                    .map(|(p1, p2)| p1 - p2)
                    .collect();
                
                let crossings = count_zero_crossings(&spread);
                
                if crossings >= min_crossings {
                    qualifying_pairs.push(StockPair {
                        stock1_idx: i,
                        stock2_idx: j,
                        distance,
                    });
                }
            }
        }
    }
    
    // Sort by number of crossings (could extend StockPair to include this)
    qualifying_pairs.sort_by(|a, b| a.distance.partial_cmp(&b.distance).unwrap());
    qualifying_pairs
}

Учёт исторического стандартного отклонения

fn find_highsd_pairs(
    normalized_prices: &[Vec<f64>],
    top_distance_count: usize,
    min_volatility: f64
) -> Vec<StockPair> {
    let stock_count = normalized_prices.len();
    let mut all_pairs = Vec::new();
    
    for i in 0..stock_count {
        for j in (i+1)..stock_count {
            let distance = euclidean_squared_distance(&normalized_prices[i], &normalized_prices[j]);
            
            // Calculate spread volatility
            let spread: Vec<f64> = normalized_prices[i].iter()
                .zip(normalized_prices[j].iter())
                .map(|(p1, p2)| p1 - p2)
                .collect();
            
            let volatility = calculate_spread_volatility(&normalized_prices[i], &normalized_prices[j]);
            
            if volatility >= min_volatility {
                all_pairs.push(StockPair {
                    stock1_idx: i,
                    stock2_idx: j,
                    distance,
                });
            }
        }
    }
    
    // Sort by distance
    all_pairs.sort_by(|a, b| a.distance.partial_cmp(&b.distance).unwrap());
    
    // Take top N pairs with highest volatility that meet distance criteria
    all_pairs.into_iter().take(top_distance_count).collect()
}

Продвинутый подход: метод корреляции Пирсона

Реализация на Rust

fn pearson_correlation(x: &[f64], y: &[f64]) -> f64 {
    assert_eq!(x.len(), y.len(), "Arrays must have the same length");
    
    let n = x.len() as f64;
    let sum_x: f64 = x.iter().sum();
    let sum_y: f64 = y.iter().sum();
    let sum_xx: f64 = x.iter().map(|&val| val * val).sum();
    let sum_yy: f64 = y.iter().map(|&val| val * val).sum();
    let sum_xy: f64 = x.iter().zip(y.iter()).map(|(&xi, &yi)| xi * yi).sum();
    
    let numerator = n * sum_xy - sum_x * sum_y;
    let denominator = ((n * sum_xx - sum_x * sum_x) * (n * sum_yy - sum_y * sum_y)).sqrt();
    
    if denominator.abs() < f64::EPSILON {
        return 0.0;
    }
    
    numerator / denominator
}

struct PearsonPair {
    stock_idx: usize,
    comover_indices: Vec<usize>,
    correlations: Vec<f64>,
}

fn find_pearson_pairs(returns: &[Vec<f64>], top_n_comovers: usize) -> Vec<PearsonPair> {
    let stock_count = returns.len();
    let mut all_pairs = Vec::new();
    
    for i in 0..stock_count {
        let mut correlations = Vec::with_capacity(stock_count - 1);
        
        for j in 0..stock_count {
            if i == j {
                continue;
            }
            
            let correlation = pearson_correlation(&returns[i], &returns[j]).abs();
            correlations.push((j, correlation));
        }
        
        // Sort by correlation (highest first)
        correlations.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal));
        
        // Take top N comovers
        let top_comovers: Vec<(usize, f64)> = correlations.into_iter()
            .take(top_n_comovers)
            .collect();
            
        let (comover_indices, correlation_values): (Vec<usize>, Vec<f64>) = 
            top_comovers.into_iter().unzip();
            
        all_pairs.push(PearsonPair {
            stock_idx: i,
            comover_indices,
            correlations: correlation_values,
        });
    }
    
    all_pairs
}

Формирование портфеля и расчёт беты

fn calculate_beta(stock_returns: &[f64], portfolio_returns: &[f64]) -> f64 {
    let cov_xy = covariance(stock_returns, portfolio_returns);
    let var_x = variance(portfolio_returns);
    
    if var_x.abs() < f64::EPSILON {
        return 0.0;
    }
    
    cov_xy / var_x
}

fn covariance(x: &[f64], y: &[f64]) -> f64 {
    assert_eq!(x.len(), y.len());
    
    let n = x.len() as f64;
    let mean_x: f64 = x.iter().sum::<f64>() / n;
    let mean_y: f64 = y.iter().sum::<f64>() / n;
    
    let sum_cov: f64 = x.iter()
        .zip(y.iter())
        .map(|(&xi, &yi)| (xi - mean_x) * (yi - mean_y))
        .sum();
    
    sum_cov / n
}

fn variance(x: &[f64]) -> f64 {
    let n = x.len() as f64;
    let mean: f64 = x.iter().sum::<f64>() / n;
    
    let sum_var: f64 = x.iter()
        .map(|&xi| (xi - mean).powi(2))
        .sum();
    
    sum_var / n
}

Генерация торговых сигналов

Заключительный этап обоих подходов — генерация торговых сигналов на основе порогов расхождения:

enum TradingSignal {
    Long,
    Short,
    Neutral
}

struct TradePosition {
    stock1_idx: usize,
    stock2_idx: usize,
    signal: TradingSignal,
    entry_spread: f64,
    timestamp: usize,
}

fn generate_trading_signals(
    normalized_prices: &[Vec<f64>],
    pairs: &[StockPair], 
    threshold_multiplier: f64,
    volatilities: &[f64],
    current_time: usize
) -> Vec<TradePosition> {
    let mut positions = Vec::new();
    
    for (pair_idx, pair) in pairs.iter().enumerate() {
        let stock1_idx = pair.stock1_idx;
        let stock2_idx = pair.stock2_idx;
        
        // Calculate current spread
        let current_spread = normalized_prices[stock1_idx][current_time] - 
                             normalized_prices[stock2_idx][current_time];
                             
        let threshold = threshold_multiplier * volatilities[pair_idx];
        
        let signal = if current_spread > threshold {
            // Stock1 is overvalued relative to Stock2
            TradingSignal::Short
        } else if current_spread < -threshold {
            // Stock1 is undervalued relative to Stock2
            TradingSignal::Long
        } else {
            TradingSignal::Neutral
        };
        
        if signal != TradingSignal::Neutral {
            positions.push(TradePosition {
                stock1_idx,
                stock2_idx,
                signal,
                entry_spread: current_spread,
                timestamp: current_time,
            });
        }
    }
    
    positions
}

Оптимизация производительности

#[cfg(target_arch = "x86_64")]
use std::arch::x86_64::*;

#[cfg(target_arch = "x86_64")]
#[inline]
unsafe fn euclidean_distance_simd(x: &[f32], y: &[f32]) -> f32 {
    assert_eq!(x.len(), y.len());
    
    let mut sum = _mm256_setzero_ps();
    let chunks = x.len() / 8;
    
    for i in 0..chunks {
        let xi = _mm256_loadu_ps(&x[i * 8]);
        let yi = _mm256_loadu_ps(&y[i * 8]);
        
        let diff = _mm256_sub_ps(xi, yi);
        let squared = _mm256_mul_ps(diff, diff);
        sum = _mm256_add_ps(sum, squared);
    }
    
    // Handle the remaining elements
    let mut result = _mm256_reduce_add_ps(sum);
    
    for i in (chunks * 8)..x.len() {
        result += (x[i] - y[i]).powi(2);
    }
    
    result.sqrt()
}

// Helper function to sum SIMD vector
#[cfg(target_arch = "x86_64")]
#[inline(always)]
unsafe fn _mm256_reduce_add_ps(v: __m256) -> f32 {
    let hilow = _mm256_extractf128_ps(v, 1);
    let low = _mm256_castps256_ps128(v);
    let sum128 = _mm_add_ps(hilow, low);
    
    let hi64 = _mm_extractf128_si128(_mm_castps_si128(sum128), 1);
    let low64 = _mm_castps_si128(sum128);
    let sum64 = _mm_add_ps(_mm_castsi128_ps(hi64), _mm_castsi128_ps(low64));
    
    _mm_cvtss_f32(_mm_hadd_ps(sum64, sum64))
}

use tokio::task;
use futures::future::join_all;

async fn process_pairs_async(
    normalized_prices: &[Vec<f64>],
    stock_count: usize,
    chunk_size: usize
) -> Vec<StockPair> {
    let mut tasks = Vec::new();
    
    // Split work into chunks
    let chunks = (stock_count + chunk_size - 1) / chunk_size;
    
    for chunk in 0..chunks {
        let start = chunk * chunk_size;
        let end = std::cmp::min((chunk + 1) * chunk_size, stock_count);
        
        let prices_clone = normalized_prices.to_vec();
        
        let task = task::spawn(async move {
            let mut pairs = Vec::new();
            
            for i in start..end {
                for j in (i+1)..stock_count {
                    let distance = euclidean_squared_distance(&prices_clone[i], &prices_clone[j]);
                    
                    pairs.push(StockPair {
                        stock1_idx: i,
                        stock2_idx: j,
                        distance,
                    });
                }
            }
            
            pairs
        });
        
        tasks.push(task);
    }
    
    // Await all tasks and combine results
    let results = join_all(tasks).await;
    
    let mut all_pairs = Vec::new();
    for result in results {
        if let Ok(pairs) = result {
            all_pairs.extend(pairs);
        }
    }
    
    // Sort by distance
    all_pairs.sort_by(|a, b| a.distance.partial_cmp(&b.distance).unwrap_or(std::cmp::Ordering::Equal));
    all_pairs
}

Тестирование реализации стратегии

Для оценки реализации необходима соответствующая инфраструктура тестирования:

#[cfg(test)]
mod tests {
    use super::*;
    
    #[test]
    fn test_normalization() {
        let prices = vec![10.0, 15.0, 12.0, 18.0, 20.0];
        let normalized = min_max_normalize(&prices);
        
        let expected = vec![0.0, 0.5, 0.2, 0.8, 1.0];
        
        for (a, b) in normalized.iter().zip(expected.iter()) {
            assert!((a - b).abs() < 0.001);
        }
    }
    
    #[test]
    fn test_euclidean_distance() {
        let x = vec![0.1, 0.2, 0.3, 0.4, 0.5];
        let y = vec![0.15, 0.22, 0.35, 0.38, 0.53];
        
        let distance = euclidean_squared_distance(&x, &y);
        let expected = 0.0049; // Calculated manually
        
        assert!((distance - expected).abs() < 0.0001);
    }
    
    #[test]
    fn test_pearson_correlation() {
        let x = vec![1.0, 2.0, 3.0, 4.0, 5.0];
        let y = vec![5.0, 4.0, 3.0, 2.0, 1.0];
        
        let corr = pearson_correlation(&x, &y);
        let expected = -1.0; // Perfect negative correlation
        
        assert!((corr - expected).abs() < 0.0001);
    }
    
    // Integration tests would be implemented in tests/ directory
}

Заключение

При реализации системы парного трейдинга важно учитывать:

Баланс между простотой (базовый подход) и статистической мощностью (подход Пирсона)
Необходимые вычислительные ресурсы для анализа большого числа пар
Транзакционные издержки, которые могут существенно влиять на прибыльность[3]
Необходимость постоянного мониторинга и перекалибровки пар

Цитирование

@software{soloviov2025distanceapproach,
  author = {Soloviov, Eugen},
  title = {Distance Approach in Pairs Trading: Implementation and Analysis with Rust},
  year = {2025},
  url = {https://marketmaker.cc/en/blog/post/distance-approach-pairs-trading},
  version = {0.1.0},
  description = {A comprehensive analysis of basic and advanced Distance Approach methodologies for pairs trading, with practical implementations in Rust tailored for high-frequency traders and algorithmic developers.}
}

Теоретические основы дистанционного подхода

Математическая основа

Базовая реализация дистанционного подхода

Нормализация данных

Поиск ближайших пар

Расчёт исторической волатильности

Продвинутые методы отбора

Фильтрация по отраслевым группам

Метод нулевых пересечений

Учёт исторического стандартного отклонения

Продвинутый подход: метод корреляции Пирсона

Реализация на Rust

Формирование портфеля и расчёт беты

Генерация торговых сигналов

Оптимизация производительности

Тестирование реализации стратегии

Заключение

Цитирование

Источники

Читайте также

Очередь внутри стенки: анализ позиции ордера в плотности ордербука

PnL по активному времени: метрика, которая меняет ранжирование стратегий

Adaptive drill-down: бэктест с переменной гранулярностью от минут до сырых сделок

Будьте в курсе событий

Теоретические основы дистанционного подхода

Математическая основа

Базовая реализация дистанционного подхода

Нормализация данных

Поиск ближайших пар

Расчёт исторической волатильности

Продвинутые методы отбора

Фильтрация по отраслевым группам

Метод нулевых пересечений

Учёт исторического стандартного отклонения

Продвинутый подход: метод корреляции Пирсона

Реализация на Rust

Формирование портфеля и расчёт беты

Генерация торговых сигналов

Оптимизация производительности

Тестирование реализации стратегии

Заключение

Цитирование

Источники