免责声明：本文提供的信息仅用于教育和参考目的，不构成财务、投资或交易建议。加密货币交易涉及重大损失风险。
MarketMaker.cc Team
量化研究与策略
MarketMaker.cc Team
量化研究与策略
距离法在配对交易中因其简洁高效而备受青睐。这一技术通过统计度量识别资产对，并基于价格关系的背离与收敛进行交易。本文全面分析了基础与高级距离法方法，并为高频交易者、算法开发者、数学家和程序员提供了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]。
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（单指令多数据）指令可显著加速距离计算：
#[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和并发处理优化下。统计严谨性与高效实现的结合，为算法交易者打造了强大工具箱。
在实现配对交易系统时，应考虑：
将距离法与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.} }