Skip to content

momentum_volatility_state

from quantmaster.features.momentum import momentum_volatility_state

df["momentum_volatility_state_20_20_60"] = momentum_volatility_state(
    df,
    mom_window=20,
    vol_window=20,
    state_window=60,
)
Source code in src/quantmaster/features/momentum.py
 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 momentum_volatility_state(
    data: pd.DataFrame | pd.Series,
    *,
    mom_window: int = 20,
    vol_window: int = 20,
    state_window: int = 60,
    price_col: str = "close",
    eps: float = 1e-12,
) -> pd.Series:
    mom_window = validate_positive_int(mom_window, name="mom_window")
    vol_window = validate_positive_int(vol_window, name="vol_window")
    state_window = validate_positive_int(state_window, name="state_window")

    if vol_window < 2:
        raise ValueError(f"vol_window must be >= 2, got {vol_window}")
    if state_window < 5:
        raise ValueError(f"state_window must be >= 5, got {state_window}")

    price = get_price_series(data, price_col=price_col).astype(float)
    price = price.where(price > 0)
    log_p = np.log(price)

    momentum = log_p.diff(mom_window)
    rets = log_p.diff()

    sigma = rets.rolling(vol_window).std(ddof=1)
    sigma_mean = sigma.rolling(state_window).mean()
    sigma_std = sigma.rolling(state_window).std(ddof=1)

    z_sigma = (sigma - sigma_mean) / sigma_std.where(sigma_std > eps)
    z_sigma = z_sigma.fillna(0.0)
    z_pos = z_sigma.clip(lower=0.0)

    weight = 1.0 / (1.0 + z_pos)
    out = momentum * weight
    out.name = f"momentum_volatility_state_{mom_window}_{vol_window}_{state_window}"
    return out