Skip to content

CORWIN–SCHULTZ SPREAD

Intuição

O Corwin–Schultz Spread (2012) estima o bid-ask spread a partir de preços high/low, explorando a diferença entre ranges de 1 dia e de 2 dias. É uma alternativa ao Roll Spread e costuma funcionar melhor em alguns cenários usando apenas OHLC.

Definição

A implementação segue a forma usual do estimador:

  • Computa beta a partir de log(high/low)^2 em dois dias consecutivos
  • Computa gamma a partir do range de 2 dias (max(high), min(low))
  • Obtém alpha e converte para spread via

S = 2*(exp(alpha)-1)/(1+exp(alpha))

O retorno da função é a média rolling de S em janela n.

Uso

from quantmaster.features.microstructure import corwin_schultz_spread

df["corwin_schultz_spread_20"] = corwin_schultz_spread(df, window=20)

API

Source code in src/quantmaster/features/microstructure.py 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def corwin_schultz_spread(
    data: pd.DataFrame,
    *,
    window: int = 20,
    high_col: str = "high",
    low_col: str = "low",
    close_col: str = "close",
) -> pd.Series:
    window = validate_positive_int(window, name="window")
    validate_columns(data, required=(high_col, low_col, close_col))

    h = pd.to_numeric(data[high_col], errors="coerce").astype(float)
    l = pd.to_numeric(data[low_col], errors="coerce").astype(float)
    c = pd.to_numeric(data[close_col], errors="coerce").astype(float)

    h = h.where(h > 0)
    l = l.where(l > 0)
    c = c.where(c > 0)

    prev_c = c.shift(1)

    gap_up = l - prev_c
    gap_down = h - prev_c

    high_adj = h.copy()
    low_adj = l.copy()

    high_adj = np.where(gap_up > 0, h - gap_up, high_adj)
    low_adj = np.where(gap_up > 0, l - gap_up, low_adj)
    high_adj = np.where(gap_down < 0, h - gap_down, high_adj)
    low_adj = np.where(gap_down < 0, l - gap_down, low_adj)

    high_adj = pd.Series(high_adj, index=data.index, dtype=float)
    low_adj = pd.Series(low_adj, index=data.index, dtype=float)
    high_adj = high_adj.where(high_adj > 0)
    low_adj = low_adj.where(low_adj > 0)

    high_2d = h.rolling(2).max()
    low_2d = l.rolling(2).min()

    beta = np.log(high_adj / low_adj).pow(2) + np.log(high_adj.shift(1) / low_adj.shift(1)).pow(2)
    gamma = np.log(high_2d / low_2d).pow(2)

    k = 3.0 - 2.0 * np.sqrt(2.0)
    alpha = ((np.sqrt(2.0 * beta) - np.sqrt(beta)) / k) - np.sqrt(gamma / k)
    alpha = alpha.clip(lower=0.0)

    spread_daily = (2.0 * (np.exp(alpha) - 1.0)) / (1.0 + np.exp(alpha))

    out = spread_daily if window == 1 else spread_daily.rolling(window).mean()
    out.name = f"corwin_schultz_spread_{window}"
    return out