Pairs Trading (Cointegration)

The Intuition

The Math

Read this as a compact model summary: what the signal sees, what it ignores, and where fragility can creep in.

log_A(t)     = log(Close_A(t))
log_B(t)     = log(Close_B(t))
hedge_ratio  = OLS coefficient of log_A on log_B
spread(t)    = log_A(t) - hedge_ratio × log_B(t)

z(t)  = (spread(t) - mean(spread[t-n:t])) / std(spread[t-n:t])

Entry:  z(t) < -entry_z  →  long spread (long A, short B)
        z(t) >  entry_z  →  short spread
Exit:   |z(t)| < exit_z  →  flat

Parameters

Source Code

def run(ticker: str, start: str, end: str, **params) -> dict:
    ticker_b = str(params.get("ticker_b", "QQQ"))
    window = int(params.get("window", 60))
    entry_z = float(params.get("entry_z", 2.0))
    exit_z = float(params.get("exit_z", 0.5))
    hedge_window = max(2, int(params.get("hedge_window", 120)))
    min_hedge_obs = max(2, int(params.get("min_hedge_obs", 60)))

    df_a = fetch_ohlcv(ticker, start, end)
    df_b = fetch_ohlcv(ticker_b, start, end)

    # Align on common dates
    common = df_a.index.intersection(df_b.index)
    if len(common) < max(window + 3, 6):
        raise ValueError("Not enough overlapping history to run the selected pair.")
    df_a = df_a.loc[common]
    df_b = df_b.loc[common]

    log_a = np.log(df_a["Close"])
    log_b = np.log(df_b["Close"])

    # Short samples cannot support the full burn-in; clamp rather than refuse.
    effective_min_obs = max(2, min(min_hedge_obs, hedge_window, len(common) // 3))

    # --- Causal hedge ratio: no beta ever sees its own bar or anything after it.
    beta = _causal_hedge_ratios(log_a, log_b, hedge_window, effective_min_obs)
    spread = log_a - beta * log_b

    # Rolling z-score of the causal spread (rolling stats are backward-looking).
    mu = spread.rolling(window).mean()
    sigma = spread.rolling(window).std()
    z = (spread - mu) / sigma.replace(0, np.nan)

    pos = pd.Series(0.0, index=df_a.index)
    for i in range(1, len(z)):
        zi = z.iloc[i]
        if pd.isna(zi):
            continue
        prev = pos.iloc[i - 1]
        if prev == 0:
            if zi < -entry_z:
                pos.iloc[i] = 1.0
            elif zi > entry_z:
                pos.iloc[i] = -1.0
        else:
            if abs(zi) < exit_z:
                pos.iloc[i] = 0.0
            else:
                pos.iloc[i] = prev

    # Time-varying leg weights follow the causal beta.
    beta_eff = beta.ffill().fillna(0.0)
    gross_notional = 1.0 + beta_eff.abs()
    weight_a = 1.0 / gross_notional
    weight_b = -beta_eff / gross_notional
    leg_a = pos * weight_a
    leg_b = pos * weight_b
    ret_a = df_a["Close"].pct_change().shift(-1).fillna(0.0)
    ret_b = df_b["Close"].pct_change().shift(-1).fillna(0.0)
    raw_pair_returns = (leg_a * ret_a + leg_b * ret_b).fillna(0.0)
    gross_turnover = leg_a.diff().abs().fillna(leg_a.abs()) + leg_b.diff().abs().fillna(leg_b.abs())

    # --- Diagnostics. The full-sample test is NOT available to the strategy.
    try:
        _, pvalue_full, _ = coint(log_a, log_b)
        pvalue_full = float(pvalue_full)
    except Exception:
        pvalue_full = float("nan")
    expanding_pvalues = _expanding_coint_pvalues(log_a, log_b, effective_min_obs)

    result = run_backtest(
        df_a,
        pos,
        ticker=ticker,
        start=start,
        end=end,
        strategy=METADATA["slug"],
        params={
            "ticker_b": ticker_b,
            "window": window,
            "entry_z": entry_z,
            "exit_z": exit_z,
            "hedge_window": hedge_window,
            "min_hedge_obs": min_hedge_obs,
        },
        include_return_series=True,
        raw_strategy_returns=raw_pair_returns,
        turnover=gross_turnover,
        trade_price_series=spread,
        **resolve_friction(params),
    )
    latest_beta = float(beta_eff.iloc[-1]) if len(beta_eff) else 0.0
    result["ticker_b"] = ticker_b
    result["pair_weights"] = {
        "ticker_a": round(float(weight_a.iloc[-1]), 4) if len(weight_a) else None,
        "ticker_b": round(float(weight_b.iloc[-1]), 4) if len(weight_b) else None,
    }
    result["hedge_ratio"] = round(latest_beta, 4)
    result["hedge_ratio_basis"] = (
        f"Trailing {hedge_window}-bar OLS re-fitted every bar on prior data only (causal)."
    )
    result["spread_latest"] = (
        round(float(spread.iloc[-1]), 4) if len(spread) and not pd.isna(spread.iloc[-1]) else None
    )
    result["zscore_latest"] = round(float(z.iloc[-1]), 4) if len(z) and not pd.isna(z.iloc[-1]) else None

    # Kept for continuity, but explicitly flagged as a non-causal descriptive stat.
    result["coint_pvalue"] = round(pvalue_full, 4) if np.isfinite(pvalue_full) else None
    result["coint_pvalue_basis"] = (
        "Full-sample Engle-Granger test. Descriptive diagnostic only: it uses the "
        "entire sample including the future, so it is a selection filter, not "
        "information the strategy had while trading."
    )
    result["coint_pvalue_is_causal"] = False
    result["coint_pvalue_expanding"] = expanding_pvalues
    result["coint_pvalue_causal"] = expanding_pvalues[0]["pvalue"] if expanding_pvalues else None

    if np.isfinite(pvalue_full) and pvalue_full > 0.05:
        result["warning"] = (
            f"Full-sample cointegration p-value {pvalue_full:.3f} > 0.05 — pair may not be "
            "cointegrated (descriptive, non-causal diagnostic)"
        )
    return result

Further Reading

• Engle, R. & Granger, C. (1987). Co-integration and Error Correction. Econometrica, 55(2), 251–276.
• Gatev, E., Goetzmann, W. & Rouwenhorst, K. (2006). Pairs Trading. Review of Financial Studies, 19(3), 797–827.
• Vidyamurthy, G. (2004). Pairs Trading: Quantitative Methods and Analysis. Wiley.
• Chan, E. (2013). Algorithmic Trading, Ch. 3. Wiley.