配对交易中目标代币的 自相关性检测同样不可或缺
有必要添加自相关性检测，但应作为辅助过滤条件，而非必需条件。
当前逻辑分析
从代码看，当前套利机会检测流程是：
if is_anomaly:
zscore_result_list = self.zscore_analysis(coin, price_data_cache)
if not zscore_result_list:
logger.info(f"❌ Z-score 计算不满足告警条件 | 币种: {coin}")
return False
# 找到绝对值最大的元素（保留原符号）
zscore_result = zscore_result_list[np.argmax(np.abs(zscore_result_list))]
self._output_results(coin, valid_results, diff_amount, zscore=zscore_result)
return True
当前只检查了：
- 相关系数（> 0.6）
- 协整检验（多周期通过）
- Z-score（多周期符号一致且超过阈值）
自相关性检测的价值
- 正自相关（趋势延续）
- 若 PURR 处于趋势中，价差可能不会立即回归
- 可延迟开仓，等待更优时机
- 负自相关（均值回归）
- 价格可能反转，加速价差回归
- 适合立即开仓
- 无显著自相关（随机游走）
- 符合均值回归假设
- 当前策略逻辑更有效
建议实现方案
在 zscore_analysis 方法返回结果后、_output_results 之前，添加自相关性检测：
def _check_autocorrelation(self, alt_prices: pd.Series, coin: str,
window: int = 50, max_lag: int = 5) -> Optional[dict]:
"""
检测PURR自身价格序列的自相关性
Args:
alt_prices: PURR价格序列
coin: 币种名称
window: 用于计算自相关的窗口大小
max_lag: 最大延迟阶数
Returns:
dict: {
'acf_values': [lag1_acf, lag2_acf, ...],
'max_acf': 最大自相关系数,
'max_lag': 最大自相关对应的延迟,
'is_positive_autocorr': 是否存在显著正自相关,
'is_negative_autocorr': 是否存在显著负自相关,
'recommendation': '开仓'/'延迟开仓'/'谨慎开仓'
}
"""
from statsmodels.tsa.stattools import acf
if len(alt_prices) < window:
return None
# 使用最近window期的收益率序列
recent_prices = alt_prices.iloc[-window:]
returns = recent_prices.pct_change().dropna()
if len(returns) < max_lag + 10:
return None
# 计算自相关函数
acf_values = acf(returns, nlags=max_lag, fft=True)
# 排除lag=0（自身相关，值为1）
acf_lags = acf_values[1:max_lag+1]
# 找出最大绝对值
max_abs_idx = np.argmax(np.abs(acf_lags))
max_acf = acf_lags[max_abs_idx]
max_lag_value = max_abs_idx + 1
# 判断显著性（使用95%置信区间，约±1.96/sqrt(n)）
n = len(returns)
significance_threshold = 1.96 / np.sqrt(n)
is_positive = max_acf > significance_threshold
is_negative = max_acf < -significance_threshold
# 给出建议
if is_positive and abs(max_acf) > 0.3:
recommendation = "延迟开仓" # 趋势延续，等待更好时机
elif is_negative and abs(max_acf) > 0.3:
recommendation = "开仓" # 均值回归，适合开仓
elif abs(max_acf) < 0.1:
recommendation = "开仓" # 无显著自相关，符合随机游走
else:
recommendation = "谨慎开仓" # 弱自相关
return {
'acf_values': acf_lags.tolist(),
'max_acf': float(max_acf),
'max_lag': int(max_lag_value),
'is_positive_autocorr': is_positive,
'is_negative_autocorr': is_negative,
'significance_threshold': float(significance_threshold),
'recommendation': recommendation
}
然后在 one_coin_analysis 中调用：
if is_anomaly:
zscore_result_list = self.zscore_analysis(coin, price_data_cache)
if not zscore_result_list:
logger.info(f"❌ Z-score 计算不满足告警条件 | 币种: {coin}")
return False
# 新增：自相关性检测
autocorr_result = None
if len(price_data_cache) > 0:
# 使用最短周期的价格数据（最敏感）
shortest_period_key = min(price_data_cache.keys(),
key=lambda x: self._timeframe_to_minutes(x[0]))
alt_prices = price_data_cache[shortest_period_key]['alt_prices']
autocorr_result = self._check_autocorrelation(alt_prices, coin)
if autocorr_result:
logger.info(
f"自相关性检测 | 币种: {coin} | "
f"最大自相关: {autocorr_result['max_acf']:.4f} (lag={autocorr_result['max_lag']}) | "
f"建议: {autocorr_result['recommendation']}"
)
# 可选：如果建议延迟开仓，可以降低信号优先级或跳过
# if autocorr_result['recommendation'] == "延迟开仓":
# logger.info(f"⚠️ 自相关检测建议延迟开仓，跳过本次信号 | 币种: {coin}")
# return False
zscore_result = zscore_result_list[np.argmax(np.abs(zscore_result_list))]
self._output_results(coin, valid_results, diff_amount,
zscore=zscore_result,
autocorr_result=autocorr_result) # 传递自相关结果
return True
注意事项
- 作为辅助指标：不要仅因自相关性就拒绝信号，应结合 Z-score 和协整结果综合判断。
- 计算成本：自相关计算较快，但需注意窗口大小。
- 参数调优：阈值（如 0.3）需根据回测结果调整。
- 多周期验证：可在多个周期上计算自相关，提高稳健性。
总结
- 有必要添加自相关性检测，用于优化开仓时机。
- 建议作为辅助过滤，而非硬性条件。
- 可先实现并记录结果，通过回测评估效果后再决定是否作为过滤条件。