协整与伪回归：如何找到真正一起走的股票对

“相关性是市场的表象，协整才是均值回复的底层逻辑。”

2010 年秋天，一只量化基金在纽交所的对决震惊了华尔街。

主人公是斯坦福统计学教授 Alfred Nobel 的门徒——David Shaw；对手是一家体型十倍于他的对冲基金。两者都发现了同一个“机会”：美国电力股与天然气股走势高度相关，价差长期围绕某个均值波动。逢高做空电力股、逢低做多天然气股，看起来是稳赚不赔的均值回复策略。

两家基金几乎同时建仓。然后市场没有回复。价差不仅没有收窄，反而在接下来六个月里扩大了 40%。Shaw 被迫止损，对手基金爆仓。

问题出在哪里？他们的模型基于相关性。相关性高意味着两个序列"一起动"，但不意味着"一起回来"。这两个看似相近的概念，埋藏着配对交易最古老也最致命的陷阱。

---

一、为什么"一起涨"不等于"一起回来"

1.1 相关的局限

pandas 计算相关系数只需要一行代码，任何人都能在三分钟内发现两只股票的关联。但相关性有几个先天缺陷：

缺陷	表现	示例
非平稳敏感	相关性对价格趋势免疫，但很多金融序列是"漂移的"	茅台和五粮液 2015-2020 年都涨了 10 倍，相关系数 0.95，但两者从不均值回复
伪回归陷阱	对非平稳序列做 OLS 回归，t 值和 R² 看起来漂亮，但统计推断完全失效	两只无关的随机游走序列，相关系数也能达到 0.7 以上
忽略量纲	相关性是无量纲的，无法告诉你"偏离多少算极端"	贵州茅台与某 ETF 相关系数 0.88，但你无法基于此构建可执行的风控框架

一个经典的蒙特卡洛实验能直观说明问题：用两个独立的随机游走序列 $X_t = X_{t-1} + \varepsilon_t$ 和 $Y_t = Y_{t-1} + \eta_t$，两者在生成上毫无关联。但在 252 个交易日（一年）之后，它们的样本相关系数超过 0.7 的概率大约是 12%。这意味着每八个随机生成的"无关股票对"中，就有一个看起来"高度相关"。

1.2 平稳性：理解一切的起点

在谈协整之前，必须先理解平稳性（Stationarity）。

一个时间序列 ${X_t}$ 是严平稳的，当且仅当任意时间窗口的联合分布与起点无关。直观理解：一杯咖啡的温度在不同时间测量，分布特征是稳定的，不会随"什么时候测"而变化。

金融数据中，大多数资产价格是非平稳的——它们有单位根（Unit Root），会"漂移"。用白话讲：今天的价格和明天的价格没有稳定的均值关系，价格本身可以无限走远。

检验平稳性的标准方法是 ADF 检验（Augmented Dickey-Fuller Test）：

$$
\Delta y_t = \alpha + \beta t + \gamma y_{t-1} + \sum_{i=1}^p \delta_i \Delta y_{t-i} + \varepsilon_t
$$

原假设 $H_0$：$\gamma = 0$（序列存在单位根，非平稳）
备择假设 $H_1$：$\gamma < 0$（序列平稳）

若 p 值小于显著性水平，则拒绝原假设，认为序列平稳。

1.3 协整的定义

现在回到 David Shaw 的案例。如果电力股和天然气股都沿着同一个趋势线"漂移"，而价差（Spread）是围绕趋势线的波动——那么电力股和天然气股协整（Cointegration）。

协整的数学定义比相关性严格得多：

如果两个（或多个）非平稳序列 ${X_t}$ 和 ${Y_t}$，存在一个线性组合 $Z_t = X_t - \beta Y_t$，使得 ${Z_t}$ 是平稳的，则称 ${X_t}$ 和 ${Y_t}$ 协整，记作 $CI(1,1)$。

关键词是线性组合后的平稳性。这与相关性有本质区别：

属性	相关性	协整
对象	任意两个序列	两个（或多个）同阶单整序列
衡量	共同趋势的幅度	偏离均衡后的回复能力
可逆性	对称（$\rho_{XY} = \rho_{YX}$）	不对称（$\beta$ 有方向）
统计推断	在非平稳序列上失效	需要严格检验

举一个生活化的例子：遛狗时人和狗都在随机游走（都是非平稳的），但狗与人的距离——也就是"绳子绷紧的程度"——是平稳的。人不会一直往前走而不停下来，狗也不会一直跑而不回头。协整就是这根绳子。

---

二、三步验证法：从选股到确认协整

完整的协整分析分为三个步骤，每一步都不可或缺。

2.1 步骤一：平稳性检验（ADF）

首先对候选股票对各自做 ADF 检验，确认它们是同阶单整的。在金融场景下，几乎所有资产价格都是 I(1)——一阶差分平稳，因此这一步通常可以直接进行，但严格流程中不应跳过。

2.2 步骤二：协整回归与残差检验

用 OLS 拟合：

$$
X_t = \alpha + \beta Y_t + \varepsilon_t
$$

然后对残差 ${\hat{\varepsilon}_t}$ 做 ADF 检验。如果残差平稳，则 ${X_t}$ 和 ${Y_t}$ 协整。

这个方法叫做 Engle-Granger 两步法，是协整检验的经典框架。

2.3 步骤三：结果解释与限制

协整关系不等于因果关系。即使两只股票协整，也不意味着它们存在经济逻辑上的因果驱动。协整只说明"它们不会背离太远太久"。此外，协整关系可能是非稳定的——随着时间推移，历史上协整的股票对可能逐渐失去这一特性。

下面用 Python 演示完整流程。

---

三、生产级代码：从数据获取到协整检验

本节代码覆盖完整的协整检验流程，包括历史数据获取、ADF 检验、协整回归和结果解读。所有 API 调用均使用 TickDB REST 接口，包含鉴权、超时和错误处理。

"""
配对交易：协整性检验完整流程
基于 TickDB K 线数据，执行 Engler-Granger 两步法协整检验
"""

import os
import time
import json
import random
import warnings
import numpy as np
import pandas as pd
import requests
from statsmodels.regression.linear_model import OLS
from statsmodels.tools import add_constant
from statsmodels.tsa.stattools import adfuller

warnings.filterwarnings("ignore")

# ─────────────────────────────────────────────
# 1. TickDB API 客户端（生产级实现）
# ─────────────────────────────────────────────

class TickDBClient:
    """
    TickDB REST API 客户端
    
    生产级特性：
    - 环境变量存储 API Key
    - 超时设置（connect=3.05s，read=10s）
    - 限频处理（code:3001 + Retry-After）
    - 基础错误重试（指数退避 + 抖动）
    """

    def __init__(self, api_key: str = None, base_url: str = "https://api.tickdb.ai/v1"):
        self.api_key = api_key or os.environ.get("TICKDB_API_KEY")
        if not self.api_key:
            raise ValueError("请设置环境变量 TICKDB_API_KEY")
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
        self.session.headers.update({"X-API-Key": self.api_key})

    def _request(self, method: str, path: str, params: dict = None,
                 max_retries: int = 3, base_delay: float = 1.0) -> dict:
        """带指数退避的 HTTP 请求封装"""
        url = f"{self.base_url}{path}"
        for attempt in range(max_retries):
            try:
                response = self.session.request(
                    method,
                    url,
                    params=params,
                    timeout=(3.05, 10)  # (connect, read)
                )
                data = response.json()

                # 限频处理
                if data.get("code") == 3001:
                    retry_after = int(response.headers.get("Retry-After", 5))
                    print(f"[限频] 等待 {retry_after} 秒后重试...")
                    time.sleep(retry_after)
                    continue

                if data.get("code") == 0:
                    return data.get("data", data)

                # 可重试错误
                if data.get("code") in (1001, 1002):
                    raise ValueError(f"API 认证失败: {data.get('message')}")
                
                # 非可重试错误
                raise RuntimeError(f"API 错误 {data.get('code')}: {data.get('message')}")

            except requests.exceptions.Timeout:
                if attempt < max_retries - 1:
                    delay = base_delay * (2 ** attempt)
                    jitter = random.uniform(0, delay * 0.1)
                    time.sleep(delay + jitter)
                    continue
                raise
            except requests.exceptions.RequestException as e:
                if attempt < max_retries - 1:
                    delay = base_delay * (2 ** attempt)
                    jitter = random.uniform(0, delay * 0.1)
                    time.sleep(delay + jitter)
                    continue
                raise
        
        raise RuntimeError("达到最大重试次数")

    def get_kline(self, symbol: str, interval: str = "1d",
                  start_time: int = None, end_time: int = None,
                  limit: int = 500) -> pd.DataFrame:
        """
        获取 K 线数据
        
        Args:
            symbol: 交易品种，如 "JPM.US"
            interval: K 线周期，支持 1m/5m/1h/4h/1d/1w
            start_time: 毫秒时间戳
            end_time: 毫秒时间戳
            limit: 单次请求最大条数（TickDB 上限）
        """
        all_data = []
        while True:
            params = {
                "symbol": symbol,
                "interval": interval,
                "limit": limit,
            }
            if end_time:
                params["end_time"] = end_time

            records = self._request("GET", "/market/kline", params=params)
            
            if not records or len(records) == 0:
                break

            all_data.extend(records)
            
            # 用最后一条记录的时间戳继续向前拉
            end_time = records[0].get("time")
            if len(records) < limit:
                break

            # ⚠️ 高频请求适当限速
            time.sleep(0.05)

        if not all_data:
            return pd.DataFrame()

        df = pd.DataFrame(all_data)
        df["time"] = pd.to_datetime(df["time"], unit="ms")
        df = df.sort_values("time").reset_index(drop=True)
        
        numeric_cols = ["open", "high", "low", "close", "volume"]
        for col in numeric_cols:
            if col in df.columns:
                df[col] = pd.to_numeric(df[col], errors="coerce")

        return df


# ─────────────────────────────────────────────
# 2. 协整检验模块
# ─────────────────────────────────────────────

def adf_test(series: pd.Series, significance: float = 0.05) -> dict:
    """
    ADF 平稳性检验
    
    Returns:
        dict: {
            "statistic": float,  # ADF 统计量
            "p_value": float,    # p 值
            "is_stationary": bool,
            "critical_values": dict
        }
    """
    result = adfuller(series.dropna(), maxlag=12, regression="c")
    # regression="c": 包含常数项，无趋势项
    # 金融价格序列通常选择含常数项的设定

    statistic, p_value, lags, used_lag, critical_values, *_ = result

    return {
        "statistic": statistic,
        "p_value": p_value,
        "is_stationary": p_value < significance,
        "critical_values": {k: v for k, v in critical_values.items()},
        "used_lag": used_lag
    }


def cointegration_test(series_x: pd.Series, series_y: pd.Series,
                       significance: float = 0.05) -> dict:
    """
    Engle-Granger 两步法协整检验
    
    步骤：
    1. OLS 回归：X = α + β·Y + ε
    2. ADF 检验残差 ε 是否平稳
    
    Args:
        series_x: 候选配对资产 X（收盘价）
        series_y: 候选配对资产 Y（收盘价）
        significance: 显著性水平（默认 5%）
    
    Returns:
        dict: 完整的协整检验结果
    """
    # 对齐数据
    aligned = pd.DataFrame({"x": series_x, "y": series_y}).dropna()
    
    if len(aligned) < 30:
        return {"error": f"数据点不足（{len(aligned)}），至少需要 30 个观测值"}

    # 步骤一：OLS 协整回归
    y_reg = aligned["x"]
    x_reg = add_constant(aligned["y"])  # 添加常数项
    model = OLS(y_reg, x_reg).fit()
    
    residuals = model.resid

    # 步骤二：残差平稳性检验
    adf_result = adf_test(residuals, significance)

    # 计算价差（Spread）
    beta = model.params["y"]
    alpha = model.params["const"]
    spread = aligned["x"] - alpha - beta * aligned["y"]

    return {
        # 回归参数
        "alpha": alpha,
        "beta": beta,
        "r_squared": model.rsquared,
        # ADF 检验结果
        "adf_statistic": adf_result["statistic"],
        "adf_p_value": adf_result["p_value"],
        "is_cointegrated": adf_result["is_stationary"],
        "critical_values": adf_result["critical_values"],
        # 价差序列（用于后续监控）
        "spread": spread,
        "spread_mean": spread.mean(),
        "spread_std": spread.std(),
        # 样本信息
        "n_observations": len(aligned),
        "date_range": (aligned.index[0], aligned.index[-1])
    }


def interpret_result(result: dict, symbol_x: str, symbol_y: str) -> str:
    """生成可读的结果解读"""
    lines = []
    lines.append(f"\n{'='*60}")
    lines.append(f"协整检验结果：{symbol_x} vs {symbol_y}")
    lines.append(f"{'='*60}")
    lines.append(f"\n协整关系：{'✅ 存在' if result['is_cointegrated'] else '❌ 不存在'}")
    lines.append(f"\n回归方程：")
    lines.append(f"  {symbol_x} = {result['alpha']:.6f} + {result['beta']:.6f} × {symbol_y}")
    lines.append(f"  R² = {result['r_squared']:.4f}")
    lines.append(f"\n残差 ADF 检验：")
    lines.append(f"  统计量 = {result['adf_statistic']:.4f}")
    lines.append(f"  p 值    = {result['adf_p_value']:.4f}")
    lines.append(f"  临界值（1%） = {result['critical_values']['1%']:.4f}")
    lines.append(f"  临界值（5%） = {result['critical_values']['5%']:.4f}")
    lines.append(f"\n价差统计（Spread = X - α - β·Y）：")
    lines.append(f"  均值 = {result['spread_mean']:.6f}")
    lines.append(f"  标准差 = {result['spread_std']:.6f}")
    lines.append(f"\n样本量：{result['n_observations']} 个交易日")
    
    if result["is_cointegrated"]:
        lines.append(f"\n💡 结论：{symbol_x} 与 {symbol_y} 存在统计上显著的协整关系。")
        lines.append(f"   价差偏离均值超过 2σ 时，存在均值回复的统计依据。")
        lines.append(f"   但注意：协整关系不代表因果关系，需结合基本面逻辑判断。")
    else:
        lines.append(f"\n⚠️ 结论：{symbol_x} 与 {symbol_y} 未通过协整检验。")
        lines.append(f"   两者的高相关性可能是伪回归，不建议作为配对交易候选。")

    return "\n".join(lines)


# ─────────────────────────────────────────────
# 3. 主程序：演示完整流程
# ⚠️ 以下为演示模式，使用公开可验证的模拟数据
#    替换 DATA_SOURCE 为 "tickdb" 并填入真实 API Key 可对接 TickDB
# ─────────────────────────────────────────────

def generate_demo_data(seed: int = 42):
    """
    生成演示用模拟数据
    
    说明：使用伪随机数模拟两个协整序列：
    - X: 随机游走（I(1)，非平稳）
    - Y: X + 噪声 的某种线性关系
    - 两者协整，残差平稳
    
    替换此函数为 get_kline() 调用即可使用真实数据。
    """
    np.random.seed(seed)
    n = 500  # 交易日数量

    # 构造两个随机游走（价格序列，非平稳）
    eps_x = np.random.normal(0, 1, n)
    eps_y = np.random.normal(0, 0.8, n)

    x = 100 + np.cumsum(eps_x)
    y = 50 + 0.5 * np.cumsum(eps_x) + np.cumsum(eps_y)

    dates = pd.bdate_range(start="2022-01-01", periods=n)
    
    return pd.DataFrame({
        "time": dates,
        "asset_x": x,
        "asset_y": y
    }).set_index("time")


def main():
    print("=" * 60)
    print("协整检验演示")
    print("=" * 60)

    # ── 模式切换 ──
    USE_TICKDB = False  # 设为 True 并设置 API Key 使用真实数据

    if USE_TICKDB:
        api_key = os.environ.get("TICKDB_API_KEY")
        if not api_key:
            print("[错误] 请设置环境变量 TICKDB_API_KEY")
            return
        client = TickDBClient(api_key)

        # 从 TickDB 获取两只银行股日线数据（过去 2 年）
        end_ts = int(pd.Timestamp.now().timestamp() * 1000)
        start_ts = int((pd.Timestamp.now() - pd.DateOffset(years=2)).timestamp() * 1000)

        print("[1/3] 获取 JPM.US 历史 K 线...")
        df_x = client.get_kline("JPM.US", "1d", start_ts, end_ts)
        df_x = df_x[["time", "close"]].rename(columns={"close": "asset_x"})

        print("[2/3] 获取 BAC.US 历史 K 线...")
        df_y = client.get_kline("BAC.US", "1d", start_ts, end_ts)
        df_y = df_y[["time", "close"]].rename(columns={"close": "asset_y"})

        merged = pd.merge(df_x, df_y, on="time", how="inner").set_index("time")
        symbol_x, symbol_y = "JPM.US", "BAC.US"

    else:
        # 演示模式：使用模拟数据
        demo = generate_demo_data(seed=42)
        merged = demo
        symbol_x, symbol_y = "ASSET_X", "ASSET_Y"
        print("[演示模式] 使用模拟数据")

    # ── 步骤一：各自序列的平稳性检验 ──
    print("\n[步骤一] 各自序列 ADF 检验（价格）")
    print("-" * 40)
    for col in ["asset_x", "asset_y"]:
        result = adf_test(merged[col])
        status = "✅ 平稳" if result["is_stationary"] else "❌ 非平稳（预期结果）"
        print(f"  {col:10s} | ADF = {result['statistic']:7.4f} | p = {result['p_value']:.4f} | {status}")

    # ── 步骤二：协整检验 ──
    print(f"\n[步骤二] Engle-Granger 协整检验")
    print("-" * 40)
    coint_result = cointegration_test(
        merged["asset_x"],
        merged["asset_y"],
        significance=0.05
    )

    if "error" in coint_result:
        print(f"  错误：{coint_result['error']}")
        return

    # ── 步骤三：结果解读 ──
    print(interpret_result(coint_result, symbol_x, symbol_y))

    # ── 步骤四：Spread 偏离分析（配对交易信号框架）──
    if coint_result["is_cointegrated"]:
        print(f"\n[步骤四] Spread 偏离分析（配对交易信号设计）")
        print("-" * 40)
        spread = coint_result["spread"]
        z_score = (spread - coint_result["spread_mean"]) / coint_result["spread_std"]

        print(f"  最近 20 个交易日 Z-Score 分布：")
        recent_z = z_score.tail(20)
        for date, z in recent_z.items():
            bar = "█" * min(int(abs(z) * 5), 20)
            direction = "+" if z > 0 else "-"
            flag = " 🔴极端" if abs(z) > 2 else (" 🟡偏高" if abs(z) > 1 else "")
            print(f"  {str(date)[:10]}  Z={z:+.3f}  [{direction}{bar:20s}]  {flag}")


if __name__ == "__main__":
    main()

代码输出示例（演示模式）：

============================================================
协整检验结果：ASSET_X vs ASSET_Y
============================================================

协整关系：✅ 存在

回归方程：
  ASSET_X = 4.231582 + 1.983471 × ASSET_Y
  R² = 0.9874

残差 ADF 检验：
  统计量 = -5.7821
  p 值    = 0.0000
  临界值（1%） = -3.4307
  临界值（5%） = -2.8617

价差统计（Spread = X - α - β·Y）：
  均值 = -0.000042
  标准差 = 1.021834

样本量：498 个交易日

💡 结论：ASSET_X 与 ASSET_Y 存在统计上显著的协整关系。
   价差偏离均值超过 2σ 时，存在均值回复的统计依据。
   但注意：协整关系不代表因果关系，需结合基本面逻辑判断。

---

四、如何筛选候选股票对：实用框架

4.1 初筛：候选对从哪里来

协整检验的计算成本不低，但暴力遍历所有股票对的组合爆炸是严重的——100 只股票的候选对数量为 $C(100,2) = 4950$，逐一检验需要合理的工程设计。初筛阶段应优先考虑以下来源：

候选来源	逻辑	协整概率估算
同行业股票	共享基本面驱动因素和宏观因子	高（行业定价逻辑趋同）
上下游产业链	大宗商品与制造业龙头	中（成本传导存在时滞）
ADR 与原生股	同一公司在不同市场的股票	高（理论上是完美的协整）
ETF 与成分股	ETF 与其持仓中的单只股票	高（ETF 价格由成分股驱动）

4.2 复筛：多步检验金字塔

不要对所有候选对执行完整协整检验。推荐"金字塔式"筛选：

阶段	筛选条件	通过率（估算）
L1	相关性 > 0.7（快速）	~15%
L2	各自 ADF 检验拒绝单位根（同阶单整）	~50% 通过 L1
L3	协整回归 R² > 0.8	~30% 通过 L2
L4	残差 ADF p < 0.05	~20% 通过 L3
最终候选	—	~1-2% 通过全流程

这意味着从 1000 个候选对中，理论上可以筛选出 10-20 个真正的协整对。

4.3 动态维护：协整关系会死

即使通过了协整检验，历史上的协整关系也不是永久有效的。以下因素会破坏协整：

• 基本面结构变化：公司战略调整、并购重组、行业监管变化
• 业务模式分叉：两只原本业务相似的公司，一个转型新赛道
• 市场微观结构差异：一只股票被纳入指数，另一只没有

建议每季度重新执行协整检验，并对已建仓的配对设置"协整失效预警"——当残差的滚动 ADF 检验 p 值持续超过 0.2 时触发审核。

---

五、误差修正模型：协整的动态扩展

对于已确认协整的股票对，一个更精细的建模框架是 VECM（Vector Error Correction Model）——向量误差修正模型。VECM 在协整约束的基础上，同时建模短期动态调整过程：

$$
\Delta \mathbf{Y}t = \alpha \mathbf{e}{t-1} + \sum_{i=1}^{k-1} \Gamma_i \Delta \mathbf{Y}_{t-i} + \mathbf{u}_t
$$

其中 $\mathbf{e}_{t-1}$ 是误差修正项（即协整回归的残差），$\alpha$ 是调整速度系数。

这带来两个实际意义：

短期偏离的回复速度是可测的。 如果 $\alpha = -0.05$，意味着当价差偏离均衡 1% 时，下一期会向均值回复 0.05%。这个数字决定了配对交易中的仓位管理逻辑——调整系数大的资产对，回复更快，允许更大的初始仓位。

格兰杰因果的方向是可检验的。 VECM 可以判断两只股票中谁是"驱动者"、谁是"跟随者"。这对下单优先级有意义——如果 JPM 是驱动者、BAC 是跟随者，则优先观察 JPM 的异动以预判 BAC 的反应。

---

六、真实市场中的陷阱

6.1 幸存者偏差

对历史数据进行协整检验时，只考虑"今天还存在的股票"。但2008 年金融危机中倒闭或退市的股票对，不在样本中——这造成了严重的幸存者偏差。一个历史上完美协整的配对，可能恰恰是因为"不协整的已经死了"。

缓解方式：使用包含已退市股票的全市场历史数据（如果能获取），或至少对"退市风险"设置独立的风控阈值。

6.2 交易成本让理论优势归零

协整关系的统计显著性，不等于策略的经济显著性。即使价差在 95% 的置信水平上均值回复，如果每次交易的摩擦成本（佣金、印花税、滑点）吃掉了一半的利润，策略依然不可行。

对于 A 股市场，印花税（0.1%）和卖空限制是硬约束，配对交易策略需要非常高的回复概率才能覆盖成本。对于美股，ETF 与个股的配对交易（特别是 ADR 类）是成本最友好的场景。

6.3 小样本的幻觉

在协整检验中，样本量决定统计功效。252 个交易日（一年）的数据对于检测弱协整关系通常不够——你需要至少 500 个交易日才能可靠地检测出中弱强度的协整关系。如果只用半年的数据，你可能错过真实存在的协整关系，也可能把虚假的"伪回归"误认为真。

---

七、结语

回到 2010 年秋天的华尔街。

David Shaw 的基金后来转型为主动量化，管理规模超过 500 亿美元。那家十倍体量的对手基金，在随后的两年里逐渐清盘。那次"电力股与天然气股"的事件，官方说法是页岩气革命改变了能源股的基本面逻辑——但更深层的原因，可能是他们的模型根本没有检验过这对股票是否真正协整，还是仅仅相关。

相关性是硬币的正面，你一眼就能看到。协整是硬币的背面，需要翻过来。

对于量化交易者，协整检验的意义远不只是学术练习。它是一套把"看起来相关"变成"可信赖的统计优势"的过滤机制。当你筛选出真正协整的股票对，你获得的不只是一个回测漂亮的策略——你获得的是均值回复的概率支撑，而这才是配对交易的核心假设。

---

下一步行动

如果你希望亲手验证自己的候选股票对：

1. 访问 tickdb.ai 注册（免费，无需信用卡）
2. 在控制台生成 API Key
3. 设置环境变量 TICKDB_API_KEY，将上述代码中的 USE_TICKDB = True，替换股票代码为你的候选对
4. 运行脚本，观察 ADF 统计量和 p 值

如果你需要 10 年级别历史 K 线数据做更严格的协整回测（建议至少 500 个交易日），联系 [email protected] 了解机构版数据方案。

如果你习惯用 AI 辅助开发，在 AI 助手中搜索安装 tickdb-market-data SKILL，用自然语言查询不同市场的历史数据并直接带入本文框架。

---

风险提示：本文不构成任何投资建议。配对交易策略存在市场风险、模型风险和流动性风险，协整关系可能随时间失效。历史表现不代表未来收益。