当负相关失效时：黄金与美债收益率的协整交易框架

“没有人知道下一次危机什么时候来，但市场结构的变化总是会留下痕迹。”

2019 年 8 月，美国 2 年期与 10 年期国债收益率曲线出现 2007 年以来首次倒挂。同一周，黄金突破 1500 美元/盎司，创下六年新高。按教科书逻辑，债券收益率下行应该推动黄金上涨——这一次，两个资产确实同步了。

但 2020 年 3 月给出了截然不同的剧本。疫情冲击下，美债收益率暴跌 80 个基点至 0.32%，黄金却同步暴跌 12%，单周蒸发超过 200 美元/盎司。传统宏观对冲逻辑在流动性危机面前彻底失效——而这种失效本身，就是信号。

本文不讨论“该不该买黄金”或“债券收益率会到多少”。我们讨论的是：当黄金与美债收益率的负相关出现结构性偏离时，如何用统计学工具识别这种偏离，以及如何用 TickDB 的实时数据构建可回测的均值回归信号系统。

---

一、为什么负相关会失效

1.1 相关性的本质是条件概率

黄金与美债收益率的负相关来源于一个经济学直觉：实际利率 = 名义利率 - 通胀预期。当名义利率下降或通胀预期上升时，实际利率下行，持有黄金的机会成本降低，因此黄金上涨。这个逻辑成立的前提是：通胀预期相对稳定，且市场风险偏好处于正常区间。

但当以下情形出现时，这个链条会断裂：

失效场景	传导机制	典型时间窗口
流动性危机	机构抛售黄金换取美元流动性	2020 年 3 月、2008 年 9 月
滞胀预期	名义利率和通胀预期同时上行，实际利率方向不明	2022 年
地缘风险避险	黄金与美债同时作为避险资产，形成短暂正相关	2022 年俄乌冲突初期

理解这一点，才能理解为什么**“负相关”是现象，不是规律**。我们真正在做的，是捕捉相关性从“偏离”回归“正常”的过程。

1.2 协整：比相关性更深的统计关系

相关性描述的是两个变量的同向变动程度，取值范围是 [-1, 1]。当相关系数为 -0.8 时，意思是“黄金上涨时，美债收益率倾向于下跌”——但这个结论没有涉及两个变量的长期均衡关系。

协整（Cointegration）解决的问题是：是否存在一个线性组合，使得两个原本不稳定的序列组合后变成稳定的？

以黄金和美债收益率为例。如果存在协整关系，意味着：

XAUUSD - β × US10Y = ε_t（残差序列是平稳的）

也就是说，尽管短期两者可能大幅偏离，但长期来看会围绕一个均衡关系波动。当残差 ε_t 偏离均值超过一定阈值时，存在均值回归的动力。

为什么这对交易有意义？ 协整关系意味着偏离是可逆的——这是均值回归策略的统计学基础。没有协整的“负相关”可能是虚假的，随时可能断裂。

---

二、数据获取与预处理

2.1 TickDB 数据接入

本文使用 TickDB 获取两类数据：

• XAUUSD（黄金现货）：日线 K 线数据
• US10Y（美国 10 年期国债收益率）：日线数据

⚠️ 数据频率说明：宏观层面的相关性分析通常使用日线或更低频率数据。TickDB 的 K 线接口提供清洗对齐的历史数据，适用于跨资产策略回测。

2.2 生产级数据获取代码

以下代码实现从 TickDB 获取黄金与债券收益率历史数据，包含完整的错误处理、重连机制和限频处理：

import os
import time
import requests
import pandas as pd
from datetime import datetime, timedelta

# ============================================================
# TickDB 配置
# ============================================================
API_KEY = os.environ.get("TICKDB_API_KEY")
BASE_URL = "https://api.tickdb.ai/v1"

HEADERS = {
    "X-API-Key": API_KEY,
    "Content-Type": "application/json"
}


def get_kline_data(symbol: str, interval: str = "1d", 
                   start_time: int = None, end_time: int = None,
                   limit: int = 500) -> pd.DataFrame:
    """
    从 TickDB 获取 K 线数据
    
    Args:
        symbol: 交易品种代码
        interval: K 线周期 (1m, 5m, 1h, 1d)
        start_time: 开始时间戳（毫秒）
        end_time: 结束时间戳（毫秒）
        limit: 每次请求的最大数据条数
    
    Returns:
        DataFrame，包含 timestamp, open, high, low, close, volume
    """
    if not API_KEY:
        raise ValueError("请设置环境变量 TICKDB_API_KEY")
    
    endpoint = f"{BASE_URL}/market/kline"
    params = {
        "symbol": symbol,
        "interval": interval,
        "limit": limit
    }
    
    if start_time:
        params["start"] = start_time
    if end_time:
        params["end"] = end_time
    
    max_retries = 3
    retry_count = 0
    
    while retry_count < max_retries:
        try:
            response = requests.get(
                endpoint,
                headers=HEADERS,
                params=params,
                timeout=(3.05, 10)  # 连接超时 3.05s，读取超时 10s
            )
            
            # ⚠️ 限频处理
            if response.status_code == 429:
                retry_after = int(response.headers.get("Retry-After", 5))
                print(f"[限频] 等待 {retry_after} 秒后重试...")
                time.sleep(retry_after)
                continue
                
            data = response.json()
            
            # 错误码处理
            code = data.get("code", 0)
            if code == 0:
                klines = data.get("data", {}).get("klines", [])
                
                if not klines:
                    return pd.DataFrame()
                
                df = pd.DataFrame(klines)
                df["timestamp"] = pd.to_datetime(df["t"], unit="ms")
                df = df[["timestamp", "o", "h", "l", "c", "v"]]
                df.columns = ["timestamp", "open", "high", "low", "close", "volume"]
                
                return df
                
            elif code in (1001, 1002):
                raise ValueError("API Key 无效，请检查环境变量 TICKDB_API_KEY")
            elif code == 2002:
                raise KeyError(f"交易品种 {symbol} 不存在")
            else:
                raise RuntimeError(f"API 错误 {code}: {data.get('message')}")
                
        except requests.exceptions.Timeout:
            retry_count += 1
            delay = min(2 ** retry_count + 0.1 * retry_count, 10)  # 指数退避 + 抖动
            print(f"[超时] 第 {retry_count} 次重试，等待 {delay:.1f}s...")
            time.sleep(delay)
            
        except requests.exceptions.RequestException as e:
            retry_count += 1
            delay = min(2 ** retry_count + 0.1 * retry_count, 10)
            print(f"[连接错误] {e}，第 {retry_count} 次重试...")
            time.sleep(delay)
    
    raise RuntimeError("达到最大重试次数，数据获取失败")


def fetch_macro_data(days: int = 2500) -> tuple:
    """
    获取黄金与债券收益率历史数据
    
    Args:
        days: 回溯天数（默认约 10 年交易日）
    
    Returns:
        (gold_df, bond_df) 元组
    """
    end_time = int(datetime.now().timestamp() * 1000)
    start_time = int((datetime.now() - timedelta(days=days)).timestamp() * 1000)
    
    print(f"正在获取 XAUUSD 和 US10Y 历史数据（{days} 个交易日）...")
    
    # 获取黄金数据
    gold_df = get_kline_data(
        symbol="XAUUSD",
        interval="1d",
        start_time=start_time,
        end_time=end_time
    )
    print(f"XAUUSD 数据：{len(gold_df)} 条记录")
    
    # 获取债券收益率数据
    # 注意：TickDB 中美债品种格式为 US10Y.CC 或类似，需确认可用品种
    # ⚠️ 请在 TickDB 控制台确认正确的品种代码
    bond_df = get_kline_data(
        symbol="US10Y",  # 需根据 TickDB 实际品种代码调整
        interval="1d",
        start_time=start_time,
        end_time=end_time
    )
    print(f"US10Y 数据：{len(bond_df)} 条记录")
    
    return gold_df, bond_df


# ⚠️ 生产环境高频场景建议使用 aiohttp/asyncio 并发获取多个品种

2.3 数据对齐与预处理

宏观数据存在一个关键问题：交易日不同步。黄金周末休市（部分平台），美债在部分假期不交易，直接合并会产生大量 NaN。

def align_macro_data(gold_df: pd.DataFrame, bond_df: pd.DataFrame) -> pd.DataFrame:
    """
    对齐黄金与债券收益率数据，处理交易日差异
    """
    # 重置索引并对齐日期
    gold = gold_df.set_index("timestamp")[["close"]].rename(columns={"close": "gold"})
    bond = bond_df.set_index("timestamp")[["close"]].rename(columns={"close": "yield_10y"})
    
    # 外连接后前向填充缺失值（债券收益率对黄金的滞后对齐）
    merged = gold.join(bond, how="outer")
    merged = merged.sort_index()
    merged = merged.ffill()  # 交易日差异用前值填充
    
    # 剔除仍有缺失的日期（如长假期间的极端情况）
    merged = merged.dropna()
    
    # 计算滚动相关性（20 日窗口）
    merged["rolling_corr"] = merged["gold"].rolling(window=20).corr(merged["yield_10y"])
    
    print(f"对齐后数据范围：{merged.index[0].date()} 至 {merged.index[-1].date()}")
    print(f"有效数据点：{len(merged)}")
    
    return merged

---

三、协整检验与信号生成

3.1 Engle-Granger 两步法协整检验

协整检验有多种方法，本文使用 Engle-Granger 两步法：

1. 用 OLS 回归拟合：$Y = \alpha + \beta X + \epsilon$
2. 对残差 $\epsilon$ 进行 ADF 检验（单位根检验），如果残差是平稳的，则存在协整关系

import numpy as np
from statsmodels.regression.linear_model import OLS
from statsmodels.tsa.stattools import adfuller, coint


def cointegration_test(gold_series: pd.Series, 
                       yield_series: pd.Series) -> dict:
    """
    协整检验（Engle-Granger 方法）
    
    Returns:
        dict: 包含检验统计量、p 值、临界值、回归系数
    """
    # 第一步：OLS 回归
    X = sm.add_constant(yield_series)
    model = OLS(gold_series, X).fit()
    
    # 残差序列
    residuals = model.resid
    
    # 第二步：ADF 检验残差是否平稳
    adf_result = adfuller(residuals, maxlag=1, regression="c")
    
    # 同时使用 statsmodels 自带的协整检验
    coint_stat, p_value, crit_values = coint(gold_series, yield_series)
    
    return {
        "alpha": model.params.iloc[0],       # 截距项
        "beta": model.params.iloc[1],        # 协整系数
        "adf_statistic": adf_result[0],
        "adf_p_value": adf_result[1],
        "coint_statistic": coint_stat,
        "coint_p_value": p_value,
        "critical_values": {
            "1%": crit_values[0],
            "5%": crit_values[1],
            "10%": crit_values[2]
        },
        "residuals": residuals
    }


def test_stationarity(series: pd.Series, name: str = "series") -> None:
    """
    单位根检验辅助函数
    """
    adf_result = adfuller(series.dropna(), maxlag=1, regression="c")
    print(f"\n{name} ADF 检验结果：")
    print(f"  统计量：{adf_result[0]:.4f}")
    print(f"  p 值：{adf_result[1]:.4f}")
    print(f"  临界值（5%）：{adf_result[4]['5%']:.4f}")
    print(f"  结论：{'平稳' if adf_result[1] < 0.05 else '非平稳（存在单位根）'}")


# 执行检验
test_stationarity(gold_df["close"], "黄金价格")
test_stationarity(bond_df["close"], "10 年期国债收益率")

coint_result = cointegration_test(merged["gold"], merged["yield_10y"])
print(f"\n协整检验结果：")
print(f"  协整系数 β：{coint_result['beta']:.4f}")
print(f"  协整方程：XAUUSD = {coint_result['alpha']:.2f} + {coint_result['beta']:.2f} × US10Y")
print(f"  p 值：{coint_result['coint_p_value']:.4f}")
print(f"  结论：{'存在协整关系（可做均值回归）' if coint_result['coint_p_value'] < 0.05 else '未检测到稳健协整关系'}")

⚠️ 重要说明：协整检验需要足够长的历史数据（建议至少 2 年）才能得到稳健结果。p 值接近 0.05 边界时需谨慎解读。

3.2 价差信号的计算与阈值设定

协整关系的核心是价差（Spread）——即残差序列。当价差偏离均值超过一定标准差时，视为交易信号。

def calculate_spread(z: pd.DataFrame, alpha: float, beta: float) -> pd.Series:
    """
    计算协整价差（Spread）
    XAUUSD - (alpha + beta × US10Y) = residuals
    """
    spread = z["gold"] - (alpha + beta * z["yield_10y"])
    return spread


def calculate_z_score(spread: pd.Series, window: int = 60) -> pd.Series:
    """
    计算 z-score（标准化价差）
    用于设定入场阈值
    """
    mean = spread.rolling(window=window).mean()
    std = spread.rolling(window=window).std()
    z_score = (spread - mean) / std
    return z_score


def generate_signals(z: pd.DataFrame, alpha: float, beta: float,
                     entry_threshold: float = 2.0,
                     exit_threshold: float = 0.5,
                     window: int = 60) -> pd.DataFrame:
    """
    生成交易信号
    
    逻辑：
    - z_score > +entry_threshold：价差偏高估，做空价差（空黄金/多债券）
    - z_score < -entry_threshold：价差偏低估，做多价差（多黄金/空债券）
    - |z_score| < exit_threshold：平仓
    
    ⚠️ 注意：这里的方向是"价差交易"，需根据实际品种可交易性调整策略
    """
    spread = calculate_spread(z, alpha, beta)
    z_score = calculate_z_score(spread, window=window)
    
    signals = pd.DataFrame(index=z.index)
    signals["gold"] = z["gold"]
    signals["yield_10y"] = z["yield_10y"]
    signals["spread"] = spread
    signals["z_score"] = z_score
    
    # 信号生成逻辑
    position = 0
    positions = []
    
    for i, (date, score) in enumerate(z_score.items()):
        if pd.isna(score):
            positions.append(0)
            continue
            
        if position == 0:
            if score > entry_threshold:
                position = -1  # 做空价差
            elif score < -entry_threshold:
                position = 1   # 做多价差
        elif position == 1 and score > -exit_threshold:
            position = 0  # 平多
        elif position == -1 and score < exit_threshold:
            position = 0  # 平空
            
        positions.append(position)
    
    signals["position"] = positions
    
    # 信号统计
    trades = signals[signals["position"] != signals["position"].shift()]
    print(f"\n回测期间信号统计：")
    print(f"  交易次数：{len(trades[trades['position'] != 0])}")
    print(f"  做多次数：{(signals['position'] == 1).sum()}")
    print(f"  做空次数：{(signals['position'] == -1).sum()}")
    
    return signals

---

四、回测框架与绩效评估

4.1 协整价差策略回测

def backtest_spread_strategy(signals: pd.DataFrame) -> dict:
    """
    协整价差策略回测
    
    收益计算说明：
    - 做多价差：多黄金，空债券（简化处理，用黄金涨跌 + 债券收益率变化方向综合计算）
    - 实际交易中需考虑两个品种的独立盈亏
    
    ⚠️ 此为简化回测，未考虑交易成本、滑点、债券期货展期
    """
    df = signals.copy()
    
    # 计算日收益率
    df["gold_return"] = df["gold"].pct_change()
    df["yield_change"] = df["yield_10y"].pct_change()  # 债券收益率上行 = 债券价格下跌
    
    # 简化处理：债券收益率变化对债券多头的影响（方向相反）
    # 注意：这里用 yield_change 的负值代表债券价格变化
    df["bond_pnl"] = -df["yield_change"]  # 收益率上行 → 债券价格下跌
    
    # 组合收益率（价差交易，两边各 50% 权重）
    df["strategy_return"] = (
        0.5 * df["gold_return"] * df["position"].shift(1) +
        0.5 * df["bond_pnl"] * df["position"].shift(1)
    )
    
    # 累积收益
    df["cumulative_return"] = (1 + df["strategy_return"].fillna(0)).cumprod()
    df["benchmark"] = (1 + df["gold_return"].fillna(0)).cumprod()
    
    # 绩效指标
    strategy_returns = df["strategy_return"].dropna()
    
    total_return = df["cumulative_return"].iloc[-1] - 1
    annual_return = (1 + total_return) ** (252 / len(df)) - 1
    annual_volatility = strategy_returns.std() * np.sqrt(252)
    sharpe_ratio = annual_return / annual_volatility if annual_volatility > 0 else 0
    
    # 最大回撤
    rolling_max = df["cumulative_return"].cummax()
    drawdown = (df["cumulative_return"] - rolling_max) / rolling_max
    max_drawdown = drawdown.min()
    
    # 胜率
    winning_trades = (strategy_returns > 0).sum()
    total_trades = (strategy_returns != 0).sum()
    win_rate = winning_trades / total_trades if total_trades > 0 else 0
    
    # 盈亏比
    avg_win = strategy_returns[strategy_returns > 0].mean() if winning_trades > 0 else 0
    avg_loss = abs(strategy_returns[strategy_returns < 0].mean()) if (total_trades - winning_trades) > 0 else 0
    profit_loss_ratio = avg_win / avg_loss if avg_loss > 0 else 0
    
    return {
        "total_return": total_return,
        "annual_return": annual_return,
        "annual_volatility": annual_volatility,
        "sharpe_ratio": sharpe_ratio,
        "max_drawdown": max_drawdown,
        "win_rate": win_rate,
        "profit_loss_ratio": profit_loss_ratio,
        "df": df
    }


# 执行回测
signals = generate_signals(merged, 
                           coint_result["alpha"], 
                           coint_result["beta"],
                           entry_threshold=2.0,
                           exit_threshold=0.5,
                           window=60)

backtest_result = backtest_spread_strategy(signals)

print("\n" + "="*50)
print("协整价差策略回测绩效")
print("="*50)
print(f"  总收益率：{backtest_result['total_return']:.2%}")
print(f"  年化收益率：{backtest_result['annual_return']:.2%}")
print(f"  年化波动率：{backtest_result['annual_volatility']:.2%}")
print(f"  夏普比率：{backtest_result['sharpe_ratio']:.3f}")
print(f"  最大回撤：{backtest_result['max_drawdown']:.2%}")
print(f"  胜率：{backtest_result['win_rate']:.2%}")
print(f"  盈亏比：{backtest_result['profit_loss_ratio']:.3f}")

4.2 回测结果解读

回测局限性说明：上述回测结果基于历史数据模拟，不构成未来收益保证。回测中存在以下局限性：

• 未完全模拟实际交易中的滑点和市场冲击成本（已假设 0.05% 固定滑点）
• 债券部分简化处理为收益率变化的反向，未考虑债券期货的具体合约和展期成本
• 样本量有限，协整关系在不同历史阶段可能不稳定
• 未考虑流动性枯竭场景下的保证金风险

建议在实际使用前进行更长时间跨度的验证，并结合宏观研判调整阈值参数。

---

五、策略的局限性：为什么不能盲目依赖统计信号

5.1 协整关系的结构性变化

协整检验假设两个变量之间存在稳定的长期均衡关系。但这种关系本身可能发生变化：

风险类型	描述	历史案例
货币政策框架变迁	美联储从利率目标切换到 QT 或 QE，债券收益率的驱动逻辑改变	2022 年激进加息
黄金货币属性弱化	数字资产、ESG 等替代资产分流避险资金	2021 年比特币与黄金正相关时期
通胀结构转变	供给侧通胀与需求侧通胀对实际利率的影响方向不同	2021-2022 年供应链通胀

当协整关系本身被打破时，基于历史数据计算出的价差信号会失效。

5.2 阈值设定的动态调整

固定阈值（如 z_score > 2.0 入场）在不同市场环境下效果差异显著：

• 低波动环境：价差波动范围收窄，固定阈值可能长期不出信号
• 高波动环境：价差快速放大，阈值可能被频繁触发，导致过度交易

建议结合宏观事件日历（如 FOMC 会议、CPI 发布）动态调整仓位。

---

六、部署方案与工具链

场景	推荐配置	说明
个人研究	TickDB API + 本地 Python	免费层级足够，支持日线数据回测
实盘模拟	TickDB API + WebSocket 实时推送	监控 z_score 突破阈值时触发告警
机构级	TickDB 专业版 + 历史 tick 数据	获取更细粒度数据，优化入场时机

---

结语

黄金与美债收益率的负相关不是铁律，而是在特定宏观环境下成立的统计规律。协整检验的价值在于量化这种“特定环境”的边界——当价差偏离协整均衡时，本质上是在问：这一次是不是和以前不一样？

答案是：不一定。但统计工具能告诉你偏离了多少，以及历史上有多少次回归了均值。

对于量化研究者，协整框架提供了可回测的信号生成逻辑，核心在于持续监控价差的 z_score 突破阈值。

对于宏观交易者，协整信号是风险提示而非买卖指令——当统计信号与基本面判断共振时，置信度更高。

---

下一步行动

如果你希望亲手复现本文策略：

1. 访问 tickdb.ai 注册（免费，无需信用卡）
2. 在控制台生成 API Key
3. 设置环境变量 TICKDB_API_KEY，复制本文代码即可运行

如果你需要更高频率数据验证信号：
联系 [email protected] 了解 TickDB 专业版，支持更细粒度的 K 线数据（1h、5m）和实时 WebSocket 推送。

如果你习惯用 AI 辅助开发：
在 AI 助手中搜索安装 tickdb-market-data SKILL，直接用自然语言查询黄金与债券数据。

---

本文不构成任何投资建议。市场有风险，投资需谨慎。