Repository Public
1"""Volatility-regime filter: daily trend position, switched off when a GARCH(1,1) forecast announces high volatility."""23import numpy as np4from arch import arch_model56META = {7 "name": "garch_regime",8 "version": "1",9 "universe": ["ETH", "BTC", "LINK", "SOL"],10 "interval": "1h",11 "lookback": 4320,12 "params": {"trend_days": 30, "vol_ratio_max": 1.25, "min_days": 120},13}141516def _daily_closes(df):17 """Last close of each COMPLETE UTC day (the current, unfinished day is dropped)."""18 index = df.index19 daily = df["close"].groupby(index.floor("D")).last()20 next_bar = index[-1] + (index[-1] - index[-2])21 if next_bar != next_bar.floor("D"):22 daily = daily.iloc[:-1]23 return daily.to_numpy()242526def _calm_regime(returns_pct, vol_ratio_max):27 """True if the one-day-ahead GARCH(1,1) volatility is not above vol_ratio_max x the median fitted volatility."""28 model = arch_model(returns_pct, mean="Zero", vol="GARCH", p=1, q=1, rescale=False)29 fit = model.fit(disp="off", show_warning=False)30 if fit.convergence_flag != 0:31 return False32 forecast = float(np.sqrt(fit.forecast(horizon=1, reindex=False).variance.to_numpy()[-1, 0]))33 typical = float(np.median(np.asarray(fit.conditional_volatility)))34 if not (np.isfinite(forecast) and typical > 0):35 return False36 return forecast <= vol_ratio_max * typical373839def target_weights(data, params):40 trend_days, min_days = int(params["trend_days"]), int(params["min_days"])41 vol_ratio_max = float(params["vol_ratio_max"])42 n = len(data)43 weights = {}44 for symbol, df in data.items():45 weights[symbol] = 0.046 if len(df) < 3:47 continue48 daily = _daily_closes(df)49 if len(daily) < max(min_days, trend_days) + 1:50 continue51 if not daily[-1] > daily[-trend_days:].mean():52 continue53 returns_pct = 100.0 * np.diff(np.log(daily))54 if _calm_regime(returns_pct, vol_ratio_max):55 weights[symbol] = 1.0 / n56 return weights57