协整检验过拟合问题(Claude)

SHI XIAOLONG

— 6 min read

问题总结

multi_coins3.py 中的协整检验存在5个主要过拟合风险:

1. Look-ahead Bias (前视偏差) ⚠️ 高风险

  • 位置: _calculate_cointegration_params 方法 (行365-445)
  • 问题: 使用全样本数据进行OLS回归和ADF检验,包含"未来"数据
  • 影响: 回测结果虚高,实盘表现差

2. Multiple Testing Problem (多重测试) ⚠️ 高风险

  • 位置: zscore_analysis 方法 (行1248-1319)
  • 问题: 进行6次独立检验 (3周期×2方法),无多重比较校正
  • 影响: 假阳性率 ≈ 26.5% (远高于名义的5%)

3. Parameter Overfitting (参数过拟合) ⚠️ 中等风险

  • 位置: 类常量定义 (行131-151)
  • 问题: 硬编码固定参数,可能针对历史数据优化
  • 影响: 样本外表现下降

4. Complex Signal Conditions (复杂信号条件) ⚠️ 中等风险

  • 位置: Z-score同向性检验 (行1313-1318)
  • 问题: 要求3周期完美对齐,条件过于严格
  • 影响: 样本内完美,样本外信号稀缺

5. Fixed Significance Level (固定显著性水平) ⚠️ 低风险

  • 位置: 协整判定 (行506)
  • 问题: 使用0.05阈值未经多重比较校正
  • 影响: 与问题2相关,增加假阳性

详细改进方案

方案1: 修正前视偏差 (优先级: 🔥 最高)

当前问题代码:

# _calculate_cointegration_params: 使用全样本
model.fit(log_base, log_alt)  # ❌ 包含未来数据

改进方案:

def _calculate_cointegration_params_no_lookahead(
    base_prices: pd.Series,
    alt_prices: pd.Series,
    beta_window: int = 100
) -> Optional[dict]:
    """无前视偏差的协整参数计算"""

    # 只使用 beta_window-1 个历史点进行OLS回归
    train_base = base_prices.iloc[-(beta_window):-1]
    train_alt = alt_prices.iloc[-(beta_window):-1]

    log_base_train = np.log(train_base).values.reshape(-1, 1)
    log_alt_train = np.log(train_alt).values

    # OLS回归(不包含当前点)
    model = LinearRegression()
    model.fit(log_base_train, log_alt_train)

    # 使用训练得到的参数构建包含当前点的价差序列
    log_base_full = np.log(base_prices.iloc[-beta_window:])
    log_alt_full = np.log(alt_prices.iloc[-beta_window:])
    spread = log_alt_full - (model.intercept_ + model.coef_[0] * log_base_full)

    # ADF检验也只使用历史数据(不包含当前点)
    adf_result = adfuller(spread.iloc[:-1].values, autolag='AIC')

    return {
        'alpha': model.intercept_,
        'beta': model.coef_[0],
        'spread': spread,
        'adf_pvalue': adf_result[1]
    }

方案2: 多重比较校正 (优先级: 🔥 高)

方法A: Bonferroni校正 (保守,推荐)

# 配置项
NUM_TESTS = 6  # 3周期 × 2方法
ALPHA_NOMINAL = 0.05
ALPHA_CORRECTED = ALPHA_NOMINAL / NUM_TESTS  # 0.05/6 ≈ 0.0083

# 判定逻辑
if cointegration_result['adf_pvalue'] < ALPHA_CORRECTED:  # 更严格的阈值
    logger.info("✅ 协整检验通过 (Bonferroni校正)")

方法B: Holm-Bonferroni校正 (更强大,推荐)

def holm_bonferroni_correction(p_values: list, alpha: float = 0.05) -> list:
    """
    Holm-Bonferroni逐步校正法
    返回每个p-value是否显著
    """
    n = len(p_values)
    # 排序p-values及其索引
    sorted_indices = sorted(range(n), key=lambda i: p_values[i])
    sorted_p_values = [p_values[i] for i in sorted_indices]

    # 逐步检验
    significant = [False] * n
    for k, (idx, p_val) in enumerate(zip(sorted_indices, sorted_p_values)):
        alpha_k = alpha / (n - k)  # 逐步调整阈值
        if p_val < alpha_k:
            significant[idx] = True
        else:
            break  # 后续全部拒绝

    return significant

# 使用示例
p_values = [result['adf_pvalue'] for result in cointegration_result_list]
significant_tests = holm_bonferroni_correction(p_values, alpha=0.05)
passed_count = sum(significant_tests)

if passed_count >= 2:
    logger.info(f"✅ 多重检验校正后通过: {passed_count}/6")

方案3: 参数稳健性检验 (优先级: 🔶 中)

3.1 参数敏感性分析

def parameter_sensitivity_analysis(base_prices, alt_prices, coin):
    """
    测试参数变化对结果的影响
    """
    beta_windows = [80, 100, 120]
    zscore_windows = [20, 30, 40]
    results = []

    for beta_w in beta_windows:
        for zscore_w in zscore_windows:
            result = _calculate_zscore(
                base_prices, alt_prices,
                window=zscore_w, beta_window=beta_w
            )
            results.append((beta_w, zscore_w, result))

    # 检查结果稳定性
    zscores = [r[2] for r in results if r[2] is not None]
    if len(zscores) > 0:
        mean_z = np.mean(zscores)
        std_z = np.std(zscores)
        cv = std_z / abs(mean_z) if mean_z != 0 else float('inf')

        logger.info(f"参数敏感性 | 币种: {coin} | "
                   f"Z-score均值: {mean_z:.2f} | "
                   f"标准差: {std_z:.2f} | "
                   f"变异系数: {cv:.2f}")

        # 变异系数 > 0.3 表示结果不稳定
        return cv < 0.3
    return False

3.2 自适应参数选择

def adaptive_window_selection(prices: pd.Series, target_vol: float = 0.02):
    """
    根据波动率自适应选择窗口长度

    Args:
        prices: 价格序列
        target_vol: 目标年化波动率(默认2%)

    Returns:
        optimal_window: 最优窗口长度
    """
    returns = prices.pct_change().dropna()

    # 计算滚动波动率,找到稳定窗口
    for window in range(50, 200, 10):
        if len(returns) < window:
            continue
        vol = returns.iloc[-window:].std() * np.sqrt(252)  # 年化波动率
        if abs(vol - target_vol) < 0.005:  # 接近目标波动率
            return window

    return 100  # 默认值

方案4: 样本外验证框架 (优先级: 🔥 最高)

Walk-Forward Analysis (滚动验证)

def walk_forward_validation(
    analyzer: DelayCorrelationAnalyzer,
    coin: str,
    train_period: str = '90d',
    test_period: str = '30d',
    n_splits: int = 4
):
    """
    滚动窗口样本外验证

    流程:
    1. 训练期: 计算协整参数和阈值
    2. 测试期: 使用训练期参数生成信号
    3. 评估: 计算样本外信号准确率
    """
    results = []

    for i in range(n_splits):
        # 获取训练集和测试集
        train_end = -i * test_period_days
        train_start = train_end - train_period_days

        train_base = base_prices[train_start:train_end]
        train_alt = alt_prices[train_start:train_end]

        test_base = base_prices[train_end:train_end + test_period_days]
        test_alt = alt_prices[train_end:train_end + test_period_days]

        # 训练期: 计算协整参数
        ols_params = analyzer._calculate_cointegration_params(
            train_base, train_alt
        )

        if ols_params is None:
            continue

        # 测试期: 使用训练期参数生成信号
        test_spread = np.log(test_alt) - (
            ols_params['alpha'] + ols_params['beta'] * np.log(test_base)
        )
        test_zscore = (test_spread - test_spread.mean()) / test_spread.std()

        # 评估信号质量(均值回归检验)
        signal_returns = []
        for t in range(len(test_zscore) - 1):
            if abs(test_zscore.iloc[t]) > 2.0:  # 信号触发
                # 预期均值回归: Z-score回归0
                expected_return = -np.sign(test_zscore.iloc[t]) * (
                    test_zscore.iloc[t+1] - test_zscore.iloc[t]
                )
                signal_returns.append(expected_return)

        if signal_returns:
            hit_rate = sum(r > 0 for r in signal_returns) / len(signal_returns)
            results.append({
                'split': i,
                'hit_rate': hit_rate,
                'n_signals': len(signal_returns)
            })

    return pd.DataFrame(results)

方案5: 蒙特卡洛模拟 (优先级: 🔶 中)

测试假阳性率

def monte_carlo_false_positive_test(
    analyzer: DelayCorrelationAnalyzer,
    n_simulations: int = 1000,
    data_length: int = 500
):
    """
    蒙特卡洛模拟测试假阳性率

    在随机游走数据上运行策略,检验是否会产生虚假信号
    """
    false_positive_count = 0

    for i in range(n_simulations):
        # 生成两个独立的随机游走序列
        random_base = np.exp(np.cumsum(np.random.normal(0, 0.02, data_length)))
        random_alt = np.exp(np.cumsum(np.random.normal(0, 0.02, data_length)))

        base_series = pd.Series(random_base)
        alt_series = pd.Series(random_alt)

        # 运行协整检验
        result = analyzer._calculate_cointegration_params(
            base_series, alt_series
        )

        if result and result['adf_pvalue'] < 0.05:
            false_positive_count += 1

    false_positive_rate = false_positive_count / n_simulations
    logger.info(f"蒙特卡洛模拟 | 假阳性率: {false_positive_rate:.2%} | "
               f"预期: ≤5% | 模拟次数: {n_simulations}")

    return false_positive_rate

实施优先级建议

阶段1: 必须立即修复 (1-2天)

  1. ✅ 修正前视偏差 - 使用方案1,确保不使用"未来"数据
  2. ✅ 多重比较校正 - 使用Holm-Bonferroni方法(方案2B)

阶段2: 重要改进 (3-5天)

  1. ✅ 样本外验证 - 实施Walk-Forward Analysis(方案4)
  2. ✅ 蒙特卡洛测试 - 验证假阳性率(方案5)

阶段3: 优化增强 (1周)

  1. ⚠️ 参数稳健性 - 实施敏感性分析(方案3.1)
  2. ⚠️ 自适应参数 - 实施方案3.2

评估指标

修复前后对比指标

# 关键指标
metrics = {
    'look_ahead_bias': {
        'before': '使用全样本',
        'after': '仅使用历史数据'
    },
    'false_positive_rate': {
        'before': '~26.5% (6次独立检验)',
        'after': '<5% (多重比较校正)'
    },
    'sample_out_performance': {
        'before': '未测试',
        'after': 'Walk-forward验证 > 60%胜率'
    },
    'monte_carlo_test': {
        'before': '未测试',
        'after': '假阳性率 < 5%'
    }
}

风险提示

⚠️ 即使完成所有改进,配对交易策略仍然面临以下固有风险:

  1. 协整关系破裂: 历史协整不保证未来协整
  2. 市场微观结构: 交易成本、滑点可能吞噬利润
  3. 极端行情: 协整假设在危机时失效
  4. 执行风险: 高频信号需要低延迟执行

建议:

  • 实盘前进行至少3个月的样本外测试
  • 控制单次交易风险 < 1%
  • 设置止损机制防范协整破裂

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