协整检验存在的过拟合的问题（Cursor）- 基于波动率的自适应窗口优化方案

SHI XIAOLONG

— 5 min read

基于波动率的自适应窗口优化方案

核心设计原则

波动率-窗口反向关系：

  • 高波动期：使用短窗口（快速响应市场变化）
  • 低波动期：使用长窗口（平滑噪声，提高稳定性）

每币种独立优化：

  • 每个交易对维护独立的窗口参数
  • 基于自身历史波动率特征计算
  • 使用LRU缓存提升性能

实现步骤

1. 添加波动率计算模块

在 multi_coins3.py 中添加新的静态方法：

@staticmethod
def _calculate_volatility_metric(prices: pd.Series, lookback: int = 20) -> float:
    """
    计算价格序列的波动率指标（使用对数收益率标准差）
    
    Args:
        prices: 价格序列
        lookback: 回溯窗口（默认20期）
    
    Returns:
        年化波动率（百分比）
    """

技术细节：

  • 使用对数收益率：log_returns = np.log(prices / prices.shift(1))
  • 滚动标准差：rolling_std = log_returns.rolling(window=lookback).std()
  • 年化处理：根据timeframe转换为年化波动率（5m→105120, 1h→8760, 4h→2190）

2. 实现自适应窗口计算器

添加核心方法：

def _calculate_adaptive_windows(
    self,
    base_prices: pd.Series,
    alt_prices: pd.Series,
    timeframe: str,
    coin: str = None
) -> tuple[int, int]:
    """
    基于波动率动态计算最优窗口长度
    
    Returns:
        (adaptive_beta_window, adaptive_zscore_window)
    """

自适应规则：

# 1. 计算双币种波动率
base_vol = _calculate_volatility_metric(base_prices)
alt_vol = _calculate_volatility_metric(alt_prices)
combined_vol = (base_vol + alt_vol) / 2  # 加权平均

# 2. 波动率分位数映射（基于历史分布）
vol_percentile = 根据历史波动率计算当前分位数

# 3. 窗口长度映射函数
if vol_percentile >= 80:  # 极高波动
    beta_window = 60
    zscore_window = 20
elif vol_percentile >= 60:  # 高波动
    beta_window = 80
    zscore_window = 25
elif vol_percentile >= 40:  # 中等波动
    beta_window = 100  # 默认值
    zscore_window = 30
elif vol_percentile >= 20:  # 低波动
    beta_window = 120
    zscore_window = 35
else:  # 极低波动
    beta_window = 150
    zscore_window = 40

# 4. 窗口合理性检查
beta_window = max(50, min(200, beta_window))  # 限制范围
zscore_window = max(15, min(50, zscore_window))
zscore_window = min(zscore_window, beta_window // 2)  # 保证 ZSCORE < BETA/2

3. 添加币种级缓存机制

在 __init__ 方法中添加：

# 币种窗口参数缓存
self.adaptive_window_cache = {}  # {(coin, timeframe, period): (beta_w, zscore_w, timestamp)}
self.window_cache_ttl = 3600  # 缓存有效期1小时（避免过度缓存过时参数）

实现缓存管理方法：

def _get_adaptive_windows_cached(
    self,
    coin: str,
    base_prices: pd.Series,
    alt_prices: pd.Series,
    timeframe: str,
    period: str
) -> tuple[int, int]:
    """
    获取自适应窗口（带缓存和TTL）
    """
    cache_key = (coin, timeframe, period)
    current_time = time.time()
    
    # 检查缓存
    if cache_key in self.adaptive_window_cache:
        beta_w, zscore_w, timestamp = self.adaptive_window_cache[cache_key]
        if current_time - timestamp < self.window_cache_ttl:
            logger.debug(f"自适应窗口缓存命中 | 币种: {coin} | Beta={beta_w}, Zscore={zscore_w}")
            return beta_w, zscore_w
    
    # 计算新窗口参数
    beta_w, zscore_w = self._calculate_adaptive_windows(
        base_prices, alt_prices, timeframe, coin
    )
    
    # 更新缓存
    self.adaptive_window_cache[cache_key] = (beta_w, zscore_w, current_time)
    
    return beta_w, zscore_w

4. 修改现有方法调用链

修改点A：zscore_analysis 方法（1248行附近）

# 修改前
cointegration_status_total_period, cointegration_status_short_period, cointegration_result = self.multiple_cointegration_analysis(
    price_data['base_prices'],
    price_data['alt_prices'],
    coin=coin,
    stats_period_key=stats_period_key,
    beta_window=self.BETA_WINDOW,  # 固定值
    zscore_window=self.ZSCORE_WINDOW  # 固定值
)

# 修改后
# 获取自适应窗口参数
timeframe, period = stats_period_key
adaptive_beta_w, adaptive_zscore_w = self._get_adaptive_windows_cached(
    coin=coin,
    base_prices=price_data['base_prices'],
    alt_prices=price_data['alt_prices'],
    timeframe=timeframe,
    period=period
)

logger.info(f"自适应窗口 | 币种: {coin} | 周期: {stats_period_key} | Beta窗口: {adaptive_beta_w} | Zscore窗口: {adaptive_zscore_w}")

cointegration_status_total_period, cointegration_status_short_period, cointegration_result = self.multiple_cointegration_analysis(
    price_data['base_prices'],
    price_data['alt_prices'],
    coin=coin,
    stats_period_key=stats_period_key,
    beta_window=adaptive_beta_w,  # 自适应值
    zscore_window=adaptive_zscore_w  # 自适应值
)

修改点B：_calculate_zscore 方法（1278行附近）

保持方法签名不变，已经支持动态传入 beta_window 和 window 参数

5. 添加配置开关和回退机制

在类常量区添加：

# ========== 自适应窗口配置 ==========
ENABLE_ADAPTIVE_WINDOWS = True  # 是否启用自适应窗口
VOLATILITY_LOOKBACK = 20  # 波动率计算回溯期
ADAPTIVE_WINDOW_CACHE_TTL = 3600  # 缓存有效期（秒）

# 窗口长度约束（防止极端值）
MIN_BETA_WINDOW = 50
MAX_BETA_WINDOW = 200
MIN_ZSCORE_WINDOW = 15
MAX_ZSCORE_WINDOW = 50

# 回退机制：当数据不足或计算失败时使用固定值
FALLBACK_BETA_WINDOW = 100
FALLBACK_ZSCORE_WINDOW = 30

在 _get_adaptive_windows_cached 中添加回退逻辑：

try:
    if not self.ENABLE_ADAPTIVE_WINDOWS:
        logger.debug("自适应窗口已禁用，使用固定值")
        return self.FALLBACK_BETA_WINDOW, self.FALLBACK_ZSCORE_WINDOW
    
    # ... 正常计算逻辑 ...
    
except Exception as e:
    logger.warning(f"自适应窗口计算失败，使用回退值 | 币种: {coin} | 错误: {e}")
    return self.FALLBACK_BETA_WINDOW, self.FALLBACK_ZSCORE_WINDOW

6. 添加诊断日志和统计

在分析结束时输出窗口统计信息：

def _output_adaptive_window_statistics(self):
    """输出自适应窗口使用统计"""
    if not self.adaptive_window_cache:
        return
    
    beta_windows = [v[0] for v in self.adaptive_window_cache.values()]
    zscore_windows = [v[1] for v in self.adaptive_window_cache.values()]
    
    logger.info(
        f"自适应窗口统计 | "
        f"Beta窗口范围: [{min(beta_windows)}, {max(beta_windows)}], 均值: {np.mean(beta_windows):.1f} | "
        f"Zscore窗口范围: [{min(zscore_windows)}, {max(zscore_windows)}], 均值: {np.mean(zscore_windows):.1f}"
    )

在 run() 方法的末尾调用：

# 在 "分析完成" 日志后添加
self._output_adaptive_window_statistics()

预期效果

性能影响

  • 计算开销增加：约20-30%（符合预期）
  • 主要来源：每个币种需额外计算波动率（O(n) 复杂度）
  • 缓存机制减少重复计算

精度提升

  • 高波动币种：响应速度提升，减少滞后
  • 低波动币种：信号质量提升，减少假阳性
  • 不同市场状态：自动适配，避免参数失效

示例场景

币种A（高波动山寨币，近期波动率90分位）：
  固定窗口 → Beta=100, Zscore=30
  自适应窗口 → Beta=60, Zscore=20（更快响应）

币种B（低波动主流币，近期波动率10分位）：
  固定窗口 → Beta=100, Zscore=30
  自适应窗口 → Beta=150, Zscore=40（更稳定）

测试验证

建议完成后进行以下测试：

  1. 单元测试：验证波动率计算和窗口映射逻辑
  2. 对比测试：选择5-10个币种，对比固定窗口vs自适应窗口的信号质量
  3. 极端场景测试：数据不足、极端波动率时的回退机制

文件修改清单

  • multi_coins3.py
    • 新增方法：_calculate_volatility_metric（约30行）
    • 新增方法：_calculate_adaptive_windows（约50行）
    • 新增方法：_get_adaptive_windows_cached（约30行）
    • 新增方法：_output_adaptive_window_statistics（约15行）
    • 修改 __init__：添加缓存字典（3行）
    • 修改 zscore_analysis：替换固定窗口为自适应窗口（15行）
    • 修改 run：添加统计输出（1行）
    • 添加类常量：自适应窗口配置（约12行）

总计：约156行新增代码，15行修改

Read more

跑步的技巧（滚动落地）

“滚动落地（rolling contact / rolling foot strike）”不是一种教条式的“脚法”，而是一种 让冲击沿着整只脚、整条后链逐级传递的落地机制。 它的核心不是“你先用哪儿着地”，而是： 你的脚落地之后，冲击是不是像轮子一样滚过去，而不是像锤子一样砸下去。 这就是滚动落地的本质。 一、什么叫“滚动落地”？ 你可以把它理解成两种完全不同的落地方式： 1. 砸地（撞击式） 脚像锤子一样拍到地上： * 要么后跟先砸 * 要么前掌先戳 * 冲击集中在一个点 * 一个结构瞬间吃掉大部分载荷 结果就是： * 后跟砸 → 膝盖难受 * 前掌戳 → 前脚掌磨烂 * 都不是长跑友好模式 这叫 撞击式着地（impact strike）。 2. 滚地（滚动式） 脚像轮胎一样“滚”过地面： * 不是某一点硬砸 * 而是外侧中足先轻触 * 再向前滚到前掌 * 最后从大脚趾蹬离

By SHI XIAOLONG

AMI的优越性

世界模型（World Models）的具体例子 如下，我按类型分类，便于理解。每类都附带实际实现、演示效果和应用场景。 1. Yann LeCun / Meta 的 JEPA 系列（最直接对应“世界模型”概念） 这些是 LeCun 主张的非生成式抽象预测世界模型代表。 * I-JEPA（Image JEPA，2023） 输入一张图像，模型把不同区域（context 和 target）编码成抽象表示，然后预测 target 的表示（不在像素级别重建）。 例子：给定一张遮挡了部分物体的图片，模型能预测“被遮挡物体的大致位置和属性”，构建对物体持久性和空间关系的理解。 这是一个“原始世界模型”，能学习物理常识（如物体不会凭空消失）。 * V-JEPA / V-JEPA 2（Video JEPA，

By SHI XIAOLONG

什么是：“世界模型（World Models）”

世界模型（World Models） 是人工智能领域的一个核心概念，尤其在 Yann LeCun 等研究者推动的下一代 AI 架构中占据中心位置。它指的是 AI 系统在内部构建的对现实世界的抽象模拟或内部表示，让机器能够像人类或动物一样“理解”物理世界、预测未来、规划行动。 简单比喻 想象你闭上眼睛也能“看到”房间里的物体会如何移动、碰撞或掉落——这就是你大脑里的世界模型。AI 的世界模型就是类似的“数字孪生”（digital twin）或“内部模拟器”：它不是简单记住数据，而是学习世界的动态、因果关系和物理直觉（如重力、物体持久性、遮挡、因果等）。 为什么需要世界模型？ 当前主流的大型语言模型（LLM） 擅长处理文本（统计模式预测），但存在根本局限： * 缺乏对物理世界的真正理解 → 容易“幻觉”、无法可靠规划。 * 样本效率低 → 人类/

By SHI XIAOLONG

K线周期可配置化设计方案

K线周期可配置化设计方案 1. 背景与目标 当前 Beta 套利策略的 K 线周期硬编码为 "1h"，分散在多个文件中。需要： 1. 将 K 线周期从 1h 改为 2h 2. 提取为环境变量 BETA_ARB_KLINE_INTERVAL，使其可在 .env 中配置 2. 影响范围分析 2.1 需要修改的文件（共 6 个） 文件 硬编码位置 修改内容 src/trading/config.py BetaArbConfig dataclass 新增 kline_interval 字段，

By SHI XIAOLONG