페어 트레이딩에서의 거리 접근법: Rust를 활용한 구현과 분석

페어 트레이딩에서의 거리 접근법은 우아한 단순성과 효과성으로 큰 인기를 얻고 있습니다. 이 기법은 통계적 측정을 통해 자산 쌍을 식별하고, 가격 관계의 이탈과 수렴에 기반하여 거래합니다. 본 문서는 고빈도 트레이더, 알고리즘 개발자, 수학자, 견고한 솔루션을 추구하는 프로그래머를 위해 기본 및 고급 거리 접근법의 포괄적 분석과 Rust 실무 구현을 제공합니다.

거리 접근법 시각화: 자산 A와 B가 서로를 추적하며, 스프레드 이탈에 기반한 트레이딩 시그널(Long/Short)이 생성되는 모습

거리 접근법의 이론적 기반

거리 접근법은 자산 간 정규화된 가격 움직임에 기반한 페어 트레이딩 프레임워크를 구축합니다. 핵심적으로 유클리드 제곱 거리 측정을 사용하여 역사적으로 함께 움직이는 자산을 식별하고, 정규화된 가격 이탈이 통계적으로 유의한 임계값을 초과할 때 트레이딩 시그널을 생성합니다[2].

이 접근법은 두 가지 주요 단계로 구성됩니다:

1. 페어 형성 - 통계적으로 관련된 자산 쌍 식별
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) 명령어를 통해 거리 계산을 크게 가속화할 수 있습니다:

SIMD 가속: Rust에서 데이터 수준 병렬성을 활용하여 여러 가격 포인트를 동시 처리하며 지연 시간을 크게 단축

#[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 및 동시 처리와 같은 최적화를 통해 이러한 계산 집약적 전략의 구현에 이상적인 언어입니다. 통계적 엄밀성과 효율적 구현의 결합이 알고리즘 트레이더를 위한 강력한 도구 키트를 만들어냅니다.

페어 트레이딩 시스템을 구현할 때 몇 가지 고려사항이 있습니다:

1. 단순성(기본 접근법)과 강화된 통계적 파워(피어슨 접근법) 간의 트레이드오프
2. 대규모 페어 분석에 필요한 계산 자원
3. 수익성에 큰 영향을 미칠 수 있는 거래 비용[3]
4. 페어의 지속적인 모니터링 및 재보정 필요성

거리 접근법과 Rust의 성능 역량을 결합함으로써, 트레이더는 현대 시장이 요구하는 속도와 규모에서 운영 가능한 고효율 통계적 차익거래 시스템을 개발할 수 있습니다.

Citation

@software{soloviov2025distanceapproach,
  author = {Soloviov, Eugen},
  title = {Distance Approach in Pairs Trading: Implementation and Analysis with Rust},
  year = {2025},
  url = {https://marketmaker.cc/ko/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.}
}

References

1. Hudson Thames - Introduction to Distance Approach in Pairs Trading Part II
2. Hudson Thames - Distance Approach in Pairs Trading Part I
3. Reddit - Pairs Trading is Too Good to Be True?
4. GitHub - Kucoin Arbitrage
5. docs.rs - Euclidean Distance in geo crate
6. Simple Linear Regression in Rust
7. GitHub - correlation_rust
8. docs.rs - Cointegration in algolotl-ta
9. GitHub - trading_engine_rust
10. docs.rs - distances crate
11. Reddit - Looking for stats crate for Dickey-Fuller
12. crates.io - crypto-pair-trader
13. w3resource - Rust Structs and Enums Exercise
14. Rust Book - Test Organization
15. Design Patterns in Rust
16. GitHub - simd-euclidean
17. Rust by Example - Integration Testing
18. YouTube - Integration Testing in Rust
19. Stack Overflow - Calculate Total Distance Between Multiple Points
20. Databento - Pairs Trading Example
21. Rust std - f64 Primitive
22. Hudson & Thames - Distance Approach Documentation
23. GitHub - trading-algorithms-rust
24. docs.rs - linreg crate
25. Rust Book - References and Borrowing
26. Stack Overflow - How to Interpret adfuller Test Results
27. lib.rs - arima crate
28. Econometrics with R - Cointegration
29. DolphinDB - adfuller Function
30. docs.rs - arima crate (latest)
31. Wikipedia - Cointegration