Spread Z-Score

Intuição

O Spread Z-Score é uma base clássica de pairs trading: mede quão “esticado” está o spread entre um ativo e um benchmark em unidades de desvio-padrão, usando uma janela móvel.

Definição

Em uma janela n:

• x = log(P_asset)
• y = log(P_benchmark)
• beta = Cov(x, y) / Var(y) (estimado na janela)
• spread = x - beta * y
• z = (spread - mean(spread)) / std(spread)

Uso

from quantmaster.features.statistical import spread_zscore

z = spread_zscore(asset_df, benchmark_series, window=60)

API

Source code in src/quantmaster/features/statistical.py

def spread_zscore(
    data: pd.DataFrame,
    benchmark: pd.Series,
    *,
    window: int = 60,
    price_col: str = "close",
    log_prices: bool = True,
) -> pd.Series:
    window = validate_positive_int(window, name="window")

    asset_price = get_price_series(data, price_col=price_col).astype(float)
    bench_price = pd.to_numeric(benchmark, errors="coerce").astype(float)

    asset_price = asset_price.where(asset_price > 0)
    bench_price = bench_price.where(bench_price > 0)

    df = pd.concat([asset_price.rename("asset"), bench_price.rename("bench")], axis=1)

    out = pd.Series(np.nan, index=df.index, dtype=float)
    out.name = f"spread_zscore_{window}"

    if len(df) < window:
        return out

    if log_prices:
        x = np.log(df["asset"]).to_numpy(dtype=float)
        y = np.log(df["bench"]).to_numpy(dtype=float)
    else:
        x = df["asset"].to_numpy(dtype=float)
        y = df["bench"].to_numpy(dtype=float)

    xw = np.lib.stride_tricks.sliding_window_view(x, window_shape=window)
    yw = np.lib.stride_tricks.sliding_window_view(y, window_shape=window)

    beta = np.full(xw.shape[0], np.nan, dtype=float)
    for i in range(xw.shape[0]):
        beta[i] = _beta_from_windows(xw[i], yw[i])

    spread = x - np.concatenate([np.full(window - 1, np.nan, dtype=float), beta]) * y

    spread_s = pd.Series(spread, index=df.index)
    mu = spread_s.rolling(window).mean()
    sigma = spread_s.rolling(window).std(ddof=1)
    out = (spread_s - mu) / sigma.where(sigma > 0)
    out.name = f"spread_zscore_{window}"
    return out