← 返回文章列表 
🔀
May 21, 2025
5 分鐘閱讀
配對交易中的距離法：Rust實現與分析
#配對交易
#rust
#演算法交易
#統計套利
#距離法
Part 1 of 3 · Collection
Statistical Arbitrage & Pairs Trading
距離法在配對交易中因其簡潔高效而備受青睞。這一技術通過統計度量識別資產對，並基於價格關係的背離與收斂進行交易。本文全面分析了基礎與高階距離法方法，併為高頻交易者、演算法開發者、數學家和程式設計師提供了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
}
組合構建與Beta計算
皮爾遜方法為每隻股票構建相關性最高的組合，並計算迴歸係數：
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（單指令多資料）指令可顯著加速距離計算：
#[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.}
}
參考文獻
- Hudson Thames - Introduction to Distance Approach in Pairs Trading Part II
- Hudson Thames - Distance Approach in Pairs Trading Part I
- Reddit - Pairs Trading is Too Good to Be True?
- GitHub - Kucoin Arbitrage
- docs.rs - Euclidean Distance in geo crate
- Simple Linear Regression in Rust
- GitHub - correlation_rust
- docs.rs - Cointegration in algolotl-ta
- GitHub - trading_engine_rust
- docs.rs - distances crate
- Reddit - Looking for stats crate for Dickey-Fuller
- crates.io - crypto-pair-trader
- w3resource - Rust Structs and Enums Exercise
- Rust Book - Test Organization
- Design Patterns in Rust
- GitHub - simd-euclidean
- Rust by Example - Integration Testing
- YouTube - Integration Testing in Rust
- Stack Overflow - Calculate Total Distance Between Multiple Points
- Databento - Pairs Trading Example
- Rust std - f64 Primitive
- Hudson & Thames - Distance Approach Documentation
- GitHub - trading-algorithms-rust
- docs.rs - linreg crate
- Rust Book - References and Borrowing
- Stack Overflow - How to Interpret adfuller Test Results
- lib.rs - arima crate
- Econometrics with R - Cointegration
- DolphinDB - adfuller Function
- docs.rs - arima crate (latest)
- Wikipedia - Cointegration
免責宣告：本文提供的資訊僅用於教育和參考目的，不構成財務、投資或交易建議。加密貨幣交易涉及重大損失風險。
Authors
Eugen Soloviov
Trading-systems engineer
Trading-systems engineer building bots since 2017: cross-exchange arbitrage (connected up to 30 venues), cointegration-based pairs arbitrage across spot and futures, scalping, news and sentiment-driven strategies, trend algorithms, and portfolio management and balancing algorithms. Also builds sub-millisecond order execution, big-data warehouses, backtesting engines, AI agents, and trading interfaces (incl. open-source profitmaker.cc). Stack: JS/TS, Python, Rust/Zig/Go, DevOps, backend, frontend, architecture.
