Pairs Trading API Reference

financelib.quant.pairs.cointegration_test(series_a: pd.Series, series_b: pd.Series) -> Dict[str, float]

Test for cointegration between two price series using Engle-Granger method.

Regresses series_a on series_b and tests residuals for stationarity using the ADF test approximation.

Parameters:

Name	Type	Description	Default
series_a	Series	First price series.	required
series_b	Series	Second price series.	required

Returns:

Type	Description
Dict[str, float]	Dictionary with 'beta' (hedge ratio), 'adf_statistic',
Dict[str, float]	'is_cointegrated' (True if ADF < -2.86 at 5% level).

Source code in financelib/quant/pairs.py

def cointegration_test(
    series_a: pd.Series,
    series_b: pd.Series,
) -> Dict[str, float]:
    """Test for cointegration between two price series using Engle-Granger method.

    Regresses series_a on series_b and tests residuals for stationarity
    using the ADF test approximation.

    Args:
        series_a: First price series.
        series_b: Second price series.

    Returns:
        Dictionary with 'beta' (hedge ratio), 'adf_statistic',
        'is_cointegrated' (True if ADF < -2.86 at 5% level).
    """
    aligned = pd.concat([series_a, series_b], axis=1).dropna()
    if len(aligned) < 30:
        return {"beta": 0.0, "adf_statistic": 0.0, "is_cointegrated": False}

    y = aligned.iloc[:, 0].values
    x = aligned.iloc[:, 1].values

    # OLS: y = alpha + beta * x + epsilon
    X = np.column_stack([np.ones(len(x)), x])
    betas = np.linalg.lstsq(X, y, rcond=None)[0]
    hedge_ratio = float(betas[1])

    # Residuals (spread)
    residuals = y - betas[0] - hedge_ratio * x

    # ADF test on residuals (Dickey-Fuller)
    diff_resid = np.diff(residuals)
    lag_resid = residuals[:-1]

    X_adf = np.column_stack([np.ones(len(lag_resid)), lag_resid])
    adf_betas = np.linalg.lstsq(X_adf, diff_resid, rcond=None)[0]
    gamma = adf_betas[1]

    # t-statistic for gamma
    predictions = X_adf @ adf_betas
    sse = np.sum((diff_resid - predictions) ** 2)
    mse = sse / (len(diff_resid) - 2)
    var_gamma = mse / np.sum((lag_resid - lag_resid.mean()) ** 2)
    t_stat = gamma / np.sqrt(var_gamma) if var_gamma > 0 else 0

    # Critical value at 5% for Engle-Granger is approximately -3.37
    # but -2.86 is commonly used for rough screening
    is_cointegrated = t_stat < -2.86

    return {
        "beta": hedge_ratio,
        "adf_statistic": float(t_stat),
        "is_cointegrated": bool(is_cointegrated),
    }

financelib.quant.pairs.find_cointegrated_pairs(prices: pd.DataFrame, significance: float = -2.86) -> List[Dict[str, any]]

Scan a universe of assets to find cointegrated pairs.

Parameters:

Name	Type	Description	Default
prices	DataFrame	DataFrame of price series (columns = assets).	required
significance	float	ADF critical value threshold.	-2.86

Returns:

Type	Description
List[Dict[str, any]]	List of dicts with 'pair', 'adf_statistic', 'beta' for
List[Dict[str, any]]	cointegrated pairs, sorted by ADF statistic.

Source code in financelib/quant/pairs.py

def find_cointegrated_pairs(
    prices: pd.DataFrame,
    significance: float = -2.86,
) -> List[Dict[str, any]]:
    """Scan a universe of assets to find cointegrated pairs.

    Args:
        prices: DataFrame of price series (columns = assets).
        significance: ADF critical value threshold.

    Returns:
        List of dicts with 'pair', 'adf_statistic', 'beta' for
        cointegrated pairs, sorted by ADF statistic.
    """
    columns = prices.columns
    n = len(columns)
    pairs = []

    for i in range(n):
        for j in range(i + 1, n):
            result = cointegration_test(prices[columns[i]], prices[columns[j]])
            if result["adf_statistic"] < significance:
                pairs.append({
                    "pair": (columns[i], columns[j]),
                    "adf_statistic": result["adf_statistic"],
                    "beta": result["beta"],
                })

    pairs.sort(key=lambda x: x["adf_statistic"])
    return pairs

financelib.quant.pairs.spread_z_score(series_a: pd.Series, series_b: pd.Series, hedge_ratio: float, window: int = 20) -> pd.Series

Calculate the z-score of the spread between two cointegrated series.

spread = series_a - hedge_ratio * series_b z = (spread - mean(spread)) / std(spread)

Parameters:

Name	Type	Description	Default
series_a	Series	First price series.	required
series_b	Series	Second price series.	required
hedge_ratio	float	Hedge ratio (beta from cointegration test).	required
window	int	Rolling window for mean and std.	20

Returns:

Type	Description
Series	Z-score series of the spread.

Source code in financelib/quant/pairs.py

def spread_z_score(
    series_a: pd.Series,
    series_b: pd.Series,
    hedge_ratio: float,
    window: int = 20,
) -> pd.Series:
    """Calculate the z-score of the spread between two cointegrated series.

    spread = series_a - hedge_ratio * series_b
    z = (spread - mean(spread)) / std(spread)

    Args:
        series_a: First price series.
        series_b: Second price series.
        hedge_ratio: Hedge ratio (beta from cointegration test).
        window: Rolling window for mean and std.

    Returns:
        Z-score series of the spread.
    """
    spread = series_a - hedge_ratio * series_b
    mean = spread.rolling(window=window).mean()
    std = spread.rolling(window=window).std()
    return (spread - mean) / std

financelib.quant.pairs.pairs_backtest(series_a: pd.Series, series_b: pd.Series, hedge_ratio: float, entry_z: float = 2.0, exit_z: float = 0.5, window: int = 20, initial_capital: float = 100000) -> Dict[str, any]

Backtest a pairs trading strategy.

Enter long spread when z < -entry_z, short when z > entry_z. Exit when |z| < exit_z.

Parameters:

Name	Type	Description	Default
series_a	Series	First price series.	required
series_b	Series	Second price series.	required
hedge_ratio	float	Hedge ratio.	required
entry_z	float	Z-score threshold for entry.	2.0
exit_z	float	Z-score threshold for exit.	0.5
window	int	Rolling window for z-score.	20
initial_capital	float	Starting capital.	100000

Returns:

Type	Description
Dict[str, any]	Dictionary with backtest results.

Source code in financelib/quant/pairs.py

def pairs_backtest(
    series_a: pd.Series,
    series_b: pd.Series,
    hedge_ratio: float,
    entry_z: float = 2.0,
    exit_z: float = 0.5,
    window: int = 20,
    initial_capital: float = 100000,
) -> Dict[str, any]:
    """Backtest a pairs trading strategy.

    Enter long spread when z < -entry_z, short when z > entry_z.
    Exit when |z| < exit_z.

    Args:
        series_a: First price series.
        series_b: Second price series.
        hedge_ratio: Hedge ratio.
        entry_z: Z-score threshold for entry.
        exit_z: Z-score threshold for exit.
        window: Rolling window for z-score.
        initial_capital: Starting capital.

    Returns:
        Dictionary with backtest results.
    """
    z = spread_z_score(series_a, series_b, hedge_ratio, window).dropna()
    capital = initial_capital
    position = 0  # 1 = long spread, -1 = short spread, 0 = flat
    trades: List[Dict] = []
    entry_price_a = 0.0
    entry_price_b = 0.0

    a_prices = series_a.reindex(z.index)
    b_prices = series_b.reindex(z.index)

    for i in range(len(z)):
        z_val = z.iloc[i]
        price_a = float(a_prices.iloc[i])
        price_b = float(b_prices.iloc[i])

        if position == 0:
            if z_val < -entry_z:
                position = 1  # Long spread: buy A, sell B
                entry_price_a = price_a
                entry_price_b = price_b
            elif z_val > entry_z:
                position = -1  # Short spread: sell A, buy B
                entry_price_a = price_a
                entry_price_b = price_b
        elif position != 0 and abs(z_val) < exit_z:
            pnl = position * (
                (price_a - entry_price_a) - hedge_ratio * (price_b - entry_price_b)
            ) * (capital * 0.5 / entry_price_a)
            capital += pnl
            trades.append({
                "side": "long" if position == 1 else "short",
                "pnl": round(pnl, 2),
                "z_entry": float(z.iloc[max(0, i - 1)]),
                "z_exit": float(z_val),
            })
            position = 0

    profit = capital - initial_capital
    return {
        "initial_capital": initial_capital,
        "final_capital": round(capital, 2),
        "profit": round(profit, 2),
        "profit_pct": round(profit / initial_capital * 100, 2),
        "num_trades": len(trades),
        "trades": trades,
    }