spb/sp-wallet

1"""HAR-RV volatility forecast used for sizing and regime: trend position scaled by tomorrow's forecast volatility."""23import numpy as np45META = {6    "name": "har_rv_perp",7    "version": "1",8    "kind": "perp",9    "universe": ["BTC", "ETH", "SOL", "HYPE"],10    "interval": "1h",11    "lookback": 1200,12    "max_leverage": 2.0,13    "max_gross_exposure": 4.0,14    "params": {"day": 24, "week_days": 7, "month_days": 14, "trend_window": 120, "target_vol": 0.6,15               "max_weight": 0.5, "regime_mult": 1.5},16}1718DAYS_PER_YEAR = 365.0192021def har_forecast(close, day, week_days, month_days):22    """(forecast of the next day's realised variance, median daily realised variance of the window), or None.2324    Daily realised variance = sum of the squared one-bar log-returns of a block of `day` bars; the blocks are25    counted backwards from the newest bar. HAR (Corsi 2009): RV[t+1] = a + b RV[t] + c mean(RV[t-w+1..t])26    + d mean(RV[t-m+1..t]), fitted by least squares on the blocks of the window itself.27    """28    r2 = np.diff(np.log(close)) ** 229    days = len(r2) // day30    if day < 2 or week_days < 2 or month_days <= week_days or days < month_days + 12:31        return None32    rv = r2[len(r2) - days * day:].reshape(days, day).sum(axis=1)33    csum = np.concatenate([[0.0], np.cumsum(rv)])34    t = np.arange(month_days - 1, days)                       # blocks that have a full "month" behind them35    x = np.column_stack([np.ones(len(t)), rv[t], (csum[t + 1] - csum[t + 1 - week_days]) / week_days,36                         (csum[t + 1] - csum[t + 1 - month_days]) / month_days])37    coef = np.linalg.lstsq(x[:-1], rv[t[1:]], rcond=None)[0]  # row of block t explains block t+138    forecast = float(x[-1] @ coef)39    floor = float(rv[rv > 0].min()) if (rv > 0).any() else 0.040    return max(forecast, floor), float(np.median(rv))414243def target_weights(data, params):44    day, week_days, month_days = int(params["day"]), int(params["week_days"]), int(params["month_days"])45    trend_window = int(params["trend_window"])46    target_vol, max_weight = float(params["target_vol"]), min(float(params["max_weight"]), 1.0)47    regime_mult = float(params["regime_mult"])48    n = len(data)49    weights = {}50    for symbol, df in data.items():51        close = df["close"].to_numpy()52        weights[symbol] = 0.053        if trend_window < 2 or len(close) < trend_window + 1:54            continue55        fit = har_forecast(close, day, week_days, month_days)56        if fit is None:57            continue58        forecast, median = fit59        if not forecast > 0 or forecast > regime_mult * regime_mult * median:   # variance: the multiple is squared60            continue                                                           # turbulent regime: stand aside61        bars_per_day = 86400.0 / (df.index[-1] - df.index[-2]).total_seconds()62        vol = np.sqrt(forecast * bars_per_day / day * DAYS_PER_YEAR)           # annualised forecast volatility63        side = float(np.sign(close[-1] - close[-trend_window:].mean()))64        weights[symbol] = side * min(max_weight, target_vol / n / vol)65    return weights66