The spread between two stocks looked like a gift from the market gods.
In 2019, two semiconductor companies traded with what appeared to be a near-perfect correlation of 0.94 over a rolling 90-day window. Their price ratio hovered around 1.42, occasionally dipping to 1.38 and spiking to 1.46. The pairs trading desk at a mid-sized quant fund placed a mean-reversion bet: when the ratio compressed below 1.38, buy the numerator, short the denominator. When it expanded above 1.46, do the reverse.
The strategy lost 23% in six months.
What went wrong? The two stocks were not cointegrated. They merely appeared related because both happened to be riding the same semiconductor industry tailwind. When sector momentum turned, both stocks fell together — but the ratio did not mean-revert. It drifted. The statistical illusion of correlation had masqueraded as an exploitable relationship for years before the regime change exposed it.
This article dissects the mathematics of cointegration, explains why conventional correlation analysis leads pairs traders astray, and provides production-grade Python code to test whether two price series share a genuine long-run equilibrium.
The Correlation Trap
Understanding cointegration requires first understanding what it is not. Correlation measures the direction and strength of a linear relationship between two series. A correlation of 0.9 suggests that when stock A moves up, stock B tends to move up as well.
But correlation is silent on a critical question: do A and B return to a relationship after they diverge?
Consider three synthetic scenarios:
| Scenario | Relationship | Correlation | Cointegrated? |
|---|---|---|---|
| A and B both trend upward at 5% daily | Both drift apart | High (0.87) | No |
| A = B + noise (bounded around B) | Bounded spread | High (0.91) | Yes |
| A and B are independent random walks | No relationship | ~0 | No |
In the first scenario, two stocks can show textbook correlation while sharing no equilibrium relationship whatsoever. They are both trending, so they move in the same direction. But if you buy the spread (go long A, short B), it will not mean-revert — it will widen indefinitely. This is a spurious correlation, and it is remarkably common in equity markets.
A famous (possibly apocryphal) example: the S&P 500 index correlates strongly with the number of times the letter "e" appears in the first 10,000 digits of Pi. Both series have trends, unit roots, and structural breaks. Their correlation is mathematically real but economically meaningless.
The core insight: Correlation measures comovement. Cointegration measures error correction. If two series are cointegrated, deviations from their equilibrium relationship are temporary and self-correcting. If they are merely correlated, deviations can persist indefinitely.
The Mathematics of Cointegration
Cointegration formalizes the intuition that while individual price series are typically non-stationary (random walks), some linear combinations of them are stationary.
Step 1: Understanding Stationarity
A time series is stationary if its statistical properties — mean, variance, autocorrelation — do not depend on the time index. A random walk is non-stationary: its variance grows with time. Its expected value today equals its value yesterday plus a zero-mean innovation.
Formally, for a non-stationary series $y_t$:
$$y_t = y_{t-1} + \epsilon_t$$
where $\epsilon_t$ is white noise. The differenced series $\Delta y_t = y_t - y_{t-1}$ is stationary.
Step 2: The Cointegration Equation
If two series $x_t$ and $y_t$ are both integrated of order 1 — denoted $I(1)$, meaning they require one differencing to become stationary — then they are cointegrated if there exists a vector $\beta = (\beta_1, \beta_2)$ such that:
$$z_t = y_t - \beta_1 x_t - \beta_2 = \text{stationary}$$
The combination $z_t$ is called the spread or error term. If $z_t$ is stationary, then when $y_t$ drifts too far above $\beta_1 x_t + \beta_2$, it will tend to revert. The relationship holds in the long run even if individual series wander.
Step 3: The Engel-Granger Two-Step Method
The standard test for cointegration is the Engle-Granger two-step procedure:
Step 1: Regress one series on the other to estimate the equilibrium relationship:
$$y_t = \alpha + \beta x_t + \epsilon_t$$
Extract the residuals: $\hat{\epsilon}_t = y_t - \hat{\alpha} - \hat{\beta} x_t$Step 2: Test whether $\hat{\epsilon}_t$ is stationary using the Augmented Dickey-Fuller (ADF) test.
If the ADF test rejects the null hypothesis of a unit root in the residuals, the residuals are stationary, and the original series are cointegrated.
Step 4: The Augmented Dickey-Fuller Test
The ADF test estimates the following regression:
$$\Delta z_t = \alpha + \delta t + \gamma z_{t-1} + \sum_{i=1}^{p} \phi_i \Delta z_{t-i} + \epsilon_t$$
- $\alpha$ is a constant (drift)
- $\delta t$ is a time trend
- $\gamma$ is the coefficient of interest — if $\gamma < 0$, the series is mean-reverting
- The lag terms $\Delta z_{t-i}$ account for serial correlation
The null hypothesis is $H_0: \gamma = 0$ (unit root present). The alternative is $H_1: \gamma < 0$ (stationary).
The test statistic is compared against MacKinnon's critical values. If the test statistic is more negative than the critical value, reject $H_0$ — the series is stationary.
Step 5: The Error Correction Model
Once cointegration is established, the Error Correction Model (ECM) captures both short-run dynamics and long-run equilibrium adjustment:
$$\Delta y_t = \alpha_0 + \alpha_1 \Delta x_t - \lambda (y_{t-1} - \beta x_{t-1}) + \epsilon_t$$
The term $\lambda (y_{t-1} - \beta x_{t-1})$ is the error correction term. It pulls $y_t$ back toward equilibrium when the previous period's spread deviated from zero. The parameter $\lambda$ is the speed of adjustment — a higher value means faster mean-reversion.
Implementing Cointegration Tests in Python
The following production-grade code implements the full cointegration analysis pipeline. It includes synthetic data generation to demonstrate the difference between cointegrated and non-cointegrated series, the Engle-Granger test, the ADF test on residuals, and the Error Correction Model estimation.
import os
import time
import random
import warnings
import numpy as np
import pandas as pd
from datetime import datetime, timedelta
import statsmodels.api as sm
from statsmodels.tsa.stattools import adfuller, coint, adfuller
from statsmodels.regression.linear_model import OLS
warnings.filterwarnings("ignore")
def generate_synthetic_data(n_periods: int = 1000, seed: int = 42) -> pd.DataFrame:
"""
Generate two synthetic price series for demonstration:
- Series A: pure random walk (non-stationary, NOT cointegrated with B)
- Series B: random walk with drift (non-stationary, NOT cointegrated with A)
- Series C: A + noise that mean-reverts around A (COINTEGRATED with A)
Returns a DataFrame with columns: A, B, C, time_index.
"""
random.seed(seed)
np.random.seed(seed)
dates = [datetime(2022, 1, 1) + timedelta(days=i) for i in range(n_periods)]
# Series A: pure random walk
innovations_A = np.random.normal(0, 1, n_periods)
A = 100 + np.cumsum(innovations_A)
# Series B: random walk with positive drift (trending, not cointegrated with A)
innovations_B = np.random.normal(0.02, 1.2, n_periods) # Positive drift
B = 100 + np.cumsum(innovations_B)
# Series C: mean-reverts around A with some noise
# C = A + stationary noise
noise_C = np.random.normal(0, 0.5, n_periods)
# Apply MA(1) structure to noise for more realistic mean-reversion
noise_C = pd.Series(noise_C).rolling(3).mean().fillna(0).values
C = A + noise_C
df = pd.DataFrame({
"date": dates,
"series_A": A,
"series_B": B,
"series_C": C
})
df.set_index("date", inplace=True)
return df
def adf_test(series: pd.Series, max_lags: int = 12, regression: str = "c") -> dict:
"""
Perform the Augmented Dickey-Fuller test on a series.
Parameters:
series: Time series data
max_lags: Maximum number of lags to include (AIC-selected)
regression: Type of regression - 'c' (constant), 'ct' (constant + trend),
'n' (no constant/no trend)
Returns:
dict with test_statistic, p_value, critical_values, is_stationary, optimal_lags
"""
result = adfuller(
series.dropna(),
maxlag=max_lags,
regression=regression,
autolag="AIC"
)
test_stat = result[0]
p_value = result[1]
used_lag = result[2]
critical_values = result[4]
# MacKinnon临界值判断
is_stationary = p_value < 0.05 # 5% significance level
return {
"test_statistic": test_stat,
"p_value": p_value,
"critical_values": critical_values,
"optimal_lags": used_lag,
"is_stationary": is_stationary
}
def engle_granger_test(
series_y: pd.Series,
series_x: pd.Series,
significance_level: float = 0.05
) -> dict:
"""
Perform the Engle-Granger two-step cointegration test.
Step 1: OLS regression to find equilibrium relationship
Step 2: ADF test on residuals
Parameters:
series_y: Dependent variable (numerator in pairs trading)
series_x: Independent variable (denominator in pairs trading)
significance_level: p-value threshold for cointegration
Returns:
dict with beta (hedge ratio), alpha (intercept), residuals,
adf_results, and is_cointegrated
"""
# Step 1: OLS regression
X = sm.add_constant(series_x)
model = OLS(series_y, X).fit()
beta = model.params[series_x.name]
alpha = model.params["const"]
residuals = model.resid
# Step 2: ADF test on residuals
adf_result = adf_test(residuals, regression="c")
# Engle-Granger临界值 (MacKinnon)
# For n=1000, approximate critical values at 5% level: -3.34
# 使用statsmodels内置的协整检验
coint_result = coint(series_y, series_x, trend="c", autolag="AIC")
coint_stat, coint_pvalue, coint_crit_values = coint_result
is_cointegrated = coint_pvalue < significance_level
return {
"hedge_ratio": beta,
"intercept": alpha,
"residuals": residuals,
"adf_result": adf_result,
"coint_statistic": coint_stat,
"coint_pvalue": coint_pvalue,
"coint_critical_values": coint_crit_values,
"is_cointegrated": is_cointegrated,
"model_summary": model.summary()
}
def error_correction_model(
series_y: pd.Series,
series_x: pd.Series,
residuals: pd.Series,
lag_order: int = 1
) -> dict:
"""
Estimate an Error Correction Model for cointegrated series.
ECM: Δy_t = α_0 + α_1 * Δx_t - λ * (y_{t-1} - β * x_{t-1}) + ε_t
Parameters:
series_y: Dependent variable
series_x: Independent variable
residuals: Residuals from cointegration regression (y_{t-1} - β * x_{t-1})
lag_order: Number of lags for Δy and Δx
Returns:
dict with ECM parameters and model fit statistics
"""
df_ecm = pd.DataFrame(index=series_y.index[1:])
# 差分项
df_ecm["delta_y"] = series_y.diff().iloc[1:].values
df_ecm["delta_x"] = series_x.diff().iloc[1:].values
# 滞后差分项 (lag order)
for lag in range(1, lag_order + 1):
df_ecm[f"delta_y_lag{lag}"] = series_y.diff().shift(lag).iloc[1:].values
df_ecm[f"delta_x_lag{lag}"] = series_x.diff().shift(lag).iloc[1:].values
# 误差修正项 (t-1期的残差)
df_ecm["ec_term"] = residuals.shift(1).iloc[1:].values
# 去除NaN
df_ecm.dropna(inplace=True)
# OLS回归
X = sm.add_constant(df_ecm[["delta_x"] + [f"delta_x_lag{i}" for i in range(1, lag_order + 1)] +
["delta_y_lag1"] + ["ec_term"]])
X = sm.add_constant(df_ecm[["delta_x", "ec_term"]]) # 简化版本
y = df_ecm["delta_y"]
model = OLS(y, X).fit()
# 提取误差修正系数 (lambda)
ec_coefficient = model.params.get("ec_term", np.nan)
speed_of_adjustment = -ec_coefficient # ECM形式中λ = -系数
return {
"delta_coefficient": model.params.get("delta_x", np.nan),
"ec_coefficient": ec_coefficient, # 应该是负的
"speed_of_adjustment": speed_of_adjustment,
"model_summary": model.summary(),
"r_squared": model.rsquared,
"aic": model.aic
}
def rolling_cointegration_test(
series_y: pd.Series,
series_x: pd.Series,
window: int = 252,
step: int = 21
) -> pd.DataFrame:
"""
Perform rolling window cointegration tests to detect regime changes.
Parameters:
series_y: Dependent variable
series_x: Independent variable
window: Rolling window size in periods (default 252 = ~1 trading year)
step: Step size between windows
Returns:
DataFrame with test statistics, p-values, and cointegration status per window
"""
results = []
dates = []
for start in range(0, len(series_y) - window, step):
end = start + window
y_window = series_y.iloc[start:end]
x_window = series_x.iloc[start:end]
y_window.name = "y"
x_window.name = "x"
try:
eg_result = engle_granger_test(y_window, x_window)
results.append({
"start_date": series_y.index[start],
"end_date": series_y.index[end],
"coint_stat": eg_result["coint_statistic"],
"coint_pvalue": eg_result["coint_pvalue"],
"hedge_ratio": eg_result["hedge_ratio"],
"is_cointegrated": eg_result["is_cointegrated"]
})
dates.append(series_y.index[start])
except Exception as e:
# Handle singular matrix or convergence issues
print(f"Warning at window {start}-{end}: {e}")
continue
return pd.DataFrame(results)
def analyze_pairs(
series_y: pd.Series,
series_x: pd.Series,
pair_name: str = "Pair"
) -> None:
"""
Full cointegration analysis pipeline for a pair of series.
Prints comprehensive output for interpretation.
"""
print(f"\n{'='*60}")
print(f"Cointegration Analysis: {pair_name}")
print(f"{'='*60}")
# 1. Basic statistics
print(f"\n[1] Descriptive Statistics:")
print(f" {series_y.name}: mean={series_y.mean():.2f}, std={series_y.std():.2f}")
print(f" {series_x.name}: mean={series_x.mean():.2f}, std={series_x.std():.2f}")
# 2. Individual stationarity (should fail for price levels)
print(f"\n[2] Individual ADF Tests (price levels):")
adf_y = adf_test(series_y)
adf_x = adf_test(series_x)
print(f" {series_y.name}: stat={adf_y['test_statistic']:.4f}, p={adf_y['p_value']:.4f}")
print(f" {series_x.name}: stat={adf_x['test_statistic']:.4f}, p={adf_x['p_value']:.4f}")
print(f" -> Both non-stationary (p > 0.05): {adf_y['p_value'] > 0.05 and adf_x['p_value'] > 0.05}")
# 3. Engle-Granger cointegration test
print(f"\n[3] Engle-Granger Cointegration Test:")
eg_result = engle_granger_test(series_y, series_x)
print(f" Test statistic: {eg_result['coint_statistic']:.4f}")
print(f" P-value: {eg_result['coint_pvalue']:.4f}")
print(f" Critical values (1%, 5%, 10%): {eg_result['coint_critical_values']}")
print(f" Cointegrated at 5% level: {eg_result['is_cointegrated']}")
print(f" Hedge ratio (β): {eg_result['hedge_ratio']:.4f}")
print(f" Intercept (α): {eg_result['intercept']:.4f}")
# 4. Spread analysis
spread = eg_result["residuals"]
adf_spread = adf_test(spread)
print(f"\n[4] Spread (Residuals) Analysis:")
print(f" Spread mean: {spread.mean():.4f}")
print(f" Spread std: {spread.std():.4f}")
print(f" ADF test on spread: stat={adf_spread['test_statistic']:.4f}, p={adf_spread['p_value']:.4f}")
print(f" Spread is stationary: {adf_spread['is_stationary']}")
# 5. Error Correction Model
print(f"\n[5] Error Correction Model:")
ecm_result = error_correction_model(
series_y, series_x,
eg_result["residuals"],
lag_order=1
)
print(f" Speed of adjustment (λ): {ecm_result['speed_of_adjustment']:.4f}")
print(f" Delta coefficient (α_1): {ecm_result['delta_coefficient']:.4f}")
print(f" R-squared: {ecm_result['r_squared']:.4f}")
print(f" AIC: {ecm_result['aic']:.2f}")
if ecm_result['speed_of_adjustment'] > 0:
print(f" -> Positive λ means spread mean-reverts (as expected)")
return eg_result, ecm_result
def main():
"""
Main execution: generate data and demonstrate cointegration analysis.
"""
print("Cointegration Analysis Pipeline")
print("=" * 60)
# Generate synthetic data
df = generate_synthetic_data(n_periods=1000)
df.columns = ["series_A", "series_B", "series_C"]
# Analyze pair: A vs B (should NOT be cointegrated)
print("\n" + "="*60)
print("EXPECTED: A vs B should NOT be cointegrated")
print("(Both are independent random walks with different drifts)")
print("="*60)
eg_AB, ecm_AB = analyze_pairs(
df["series_A"],
df["series_B"],
"series_A vs series_B"
)
# Analyze pair: A vs C (SHOULD be cointegrated)
print("\n" + "="*60)
print("EXPECTED: A vs C SHOULD be cointegrated")
print("(C is A plus mean-reverting noise)")
print("="*60)
eg_AC, ecm_AC = analyze_pairs(
df["series_A"],
df["series_C"],
"series_A vs series_C"
)
# Summary table
print("\n" + "="*60)
print("SUMMARY: Cointegration Test Results")
print("="*60)
print(f"{'Pair':<30} {'Coint. Stat':<12} {'P-value':<10} {'Cointegrated?':<12}")
print(f"{'-'*30} {'-'*12} {'-'*10} {'-'*12}")
print(f"{'A vs B (no relationship)':<30} {eg_AB['coint_statistic']:<12.4f} {eg_AB['coint_pvalue']:<10.4f} {'No' if not eg_AB['is_cointegrated'] else 'Yes':<12}")
print(f"{'A vs C (noise around A)':<30} {eg_AC['coint_statistic']:<12.4f} {eg_AC['coint_pvalue']:<10.4f} {'No' if not eg_AC['is_cointegrated'] else 'Yes':<12}")
print("\nNote: With 1000 observations and 5% significance level,")
print("cointegration requires test statistic < -3.34 (critical value)")
if __name__ == "__main__":
# ⚠️ Production note: For real market data, replace synthetic data generation
# with TickDB API calls. Use GET /v1/market/kline for historical OHLCV,
# then compute close prices for cointegration analysis.
main()
⚠️ Engineering warning: The synthetic data generator uses a fixed seed for reproducibility. In production, replace this with real market data fetched from a data provider. For US equity pairs, ensure both securities have sufficient trading history and liquidity to support the strategy.
Interpreting the Test Results
When running the analysis above, you should observe a clear contrast between the two pairs:
Pair A vs B: Non-Cointegrated
| Metric | Value | Interpretation |
|---|---|---|
| ADF statistic | > -3.34 | Cannot reject unit root in residuals |
| P-value | > 0.05 | Residuals are non-stationary |
| Hedge ratio | Variable | Equilibrium relationship is unstable |
| Speed of adjustment | ~0 or negative | No force pulling the spread back |
The residuals from regressing A on B behave like a random walk. There is no equilibrium. Buying the spread when it compresses and expecting it to widen is not a valid strategy — it is a gamble on the continued correlation regime.
Pair A vs C: Cointegrated
| Metric | Value | Interpretation |
|---|---|---|
| ADF statistic | < -3.34 | Reject unit root in residuals |
| P-value | < 0.05 | Residuals are stationary |
| Hedge ratio | ~1.0 | Stable equilibrium relationship |
| Speed of adjustment | Positive | Spread mean-reverts with this half-life |
The residuals from regressing C on A are stationary. When C drifts above A by more than the noise band, it tends to revert. This is the foundation of a valid pairs trading strategy.
The Half-Life of Mean-Reversion
Once cointegration is confirmed, the speed of adjustment coefficient $\lambda$ from the ECM tells you how quickly deviations correct. The half-life of a spread deviation is:
$$\text{half-life} = \frac{\ln(2)}{\lambda}$$
For example, if $\lambda = 0.05$, the half-life is $\ln(2) / 0.05 \approx 13.86$ periods. This informs your trading horizon and position sizing.
| Half-life | Trading implication |
|---|---|
| < 5 days | Very fast mean-reversion; high turnover, narrow profit margins |
| 5–20 days | Moderate; typical for equity pairs |
| 20–60 days | Slow; transaction costs may erode edge |
| > 60 days | Requires large capital base and patience |
Regime Detection: Rolling Cointegration
A static cointegration test is insufficient for live trading. The equilibrium relationship between two securities can break down — a process known as cointegration regime change. Common triggers include:
- Regulatory changes affecting one leg of the pair
- Corporate events (mergers, spin-offs, earnings surprises)
- Structural shifts in market microstructure
- Changes in index composition affecting ETF pairs
Rolling window analysis detects these regime changes:
# Example: Rolling cointegration on SPY vs QQQ (S&P 500 vs NASDAQ ETF)
# In production, replace with TickDB kline data:
# response = requests.get(
# "https://api.tickdb.ai/v1/market/kline",
# headers={"X-API-Key": os.environ.get("TICKDB_API_KEY")},
# params={"symbol": "SPY.US", "interval": "1d", "limit": 500},
# timeout=(3.05, 10)
# )
rolling_results = rolling_cointegration_test(
spy_close_series, # Close prices from TickDB
qqq_close_series,
window=252, # 1-year rolling window
step=21 # Monthly update
)
# Flag regime changes
rolling_results["regime_change"] = rolling_results["is_cointegrated"].diff().abs() == 1
When regime_change == True, the pair has transitioned between cointegrated and non-cointegrated states. This is a signal to pause pairs trading on that ticker combination and investigate the fundamental cause.
Common Pitfalls in Cointegration Analysis
Pitfall 1: Ignoring Non-Stationarity in Individual Series
Always confirm that both series are $I(1)$ before applying cointegration tests. If both series are stationary to begin with, you are not dealing with cointegration — you are dealing with a standard regression. Run the ADF test on levels first.
Pitfall 2: In-Sample Overfitting
Finding a cointegrated pair in-sample is easy. Finding one that remains cointegrated out-of-sample is hard. Always:
- Split data into estimation and validation windows
- Require cointegration to hold in both windows
- Apply out-of-sample walk-forward validation before trading
Pitfall 3: Survivorship Bias
Backtesting pairs that "worked" historically while ignoring pairs that failed inflates performance estimates. Maintain a universe of candidate pairs and report performance across the full universe, not just the selected winners.
Pitfall 4: Transaction Costs
The profit from mean-reverting spread deviations must exceed transaction costs (bid-ask spread, commission, slippage). For liquid large-cap pairs, round-trip costs of 0.05–0.10% are realistic. For illiquid securities, costs can exceed the theoretical edge.
| Spread deviation | Gross profit | Round-trip cost | Net profit |
|---|---|---|---|
| 1% | $100 | $10 | $90 |
| 0.5% | $50 | $10 | $40 |
| 0.2% | $20 | $10 | $10 (marginal) |
| 0.1% | $10 | $10 | $0 (break-even) |
Real-World Application: Pairs Trading Workflow
For a live pairs trading implementation, the workflow integrates with TickDB as follows:
| Phase | Data source | Action |
|---|---|---|
| Candidate selection | Universe screening (sector, market cap match) | Pre-filter by fundamental similarity |
| Historical backtest | TickDB /v1/market/kline (1d, 10+ years) |
Cointegration + ECM estimation |
| Regime monitoring | TickDB /v1/market/kline/latest (real-time) |
Rolling cointegration, alert on regime change |
| Signal generation | Live depth / trades | Compute spread deviation from hedge ratio |
| Execution | Broker API | Place paired orders with limit prices |
Conclusion
Correlation is a snapshot. Cointegration is a commitment.
When two securities are cointegrated, they share a genuine long-run equilibrium that deviations from that equilibrium are self-correcting. When they are merely correlated, their comovement is a historical artifact that can evaporate without warning.
The Engel-Granger test, the ADF test on residuals, and the Error Correction Model form a complete toolkit for distinguishing real relationships from statistical illusions. Pair these tests with rolling window analysis to detect regime changes before they erode your edge.
The semiconductor pair that lost 23% in 2019 was never a pairs trading opportunity. It was a correlation mirage that survived long enough to collect capital from a desk that had not run the right test.
Next Steps
If you are building a pairs trading strategy, use TickDB's 10+ years of US equity historical kline data to perform out-of-sample cointegration validation before risking capital.
If you want to run this analysis yourself:
- Access tickdb.ai and generate an API key
- Set the
TICKDB_API_KEYenvironment variable - Replace the synthetic data generator in the code above with calls to
GET /v1/market/kline - Run the rolling cointegration analysis across your candidate universe
If you are researching microstructure for quantitative strategies, explore TickDB's depth channel documentation to understand order book dynamics that affect spread behavior around news events.
If you use AI coding assistants, search for and install the tickdb-market-data SKILL in your AI tool's marketplace for streamlined data access in your analysis notebooks.
This article does not constitute investment advice. Pairs trading involves significant risk including the risk of total loss. Backtested results do not guarantee future performance. Transaction costs and market impact can materially reduce or eliminate strategy returns.