spb/crypto-trade-bot

1"""Time-series momentum with volatility targeting (long-only spot variant)."""23from __future__ import annotations45import numpy as np6import pandas as pd78from ktrade.strategy.base import ParamSpec, Strategy, StrategyMeta, log_returns, vol_target_scale91011class TimeSeriesMomentum(Strategy):12    meta = StrategyMeta(13        name="tsmom",14        label="Time-series momentum (vol-targeted)",15        family="trend",16        hypothesis="Crypto returns exhibit medium-horizon autocorrelation (Moskowitz–Ooi–Pedersen style). Go long when the trailing "17        "lookback return is positive and the fast EMA is above the slow EMA; scale by inverse realized volatility; stay in cash otherwise. "18        "Spot-only: no shorting, so the strategy captures only the long leg of the premium.",19        experimental=True,20        allows_short=False,21        min_bars=120,22        params=[23            ParamSpec(name="lookback", type="int", default=48, min=5, max=2000, description="Bars for the trailing return signal"),24            ParamSpec(name="ema_fast", type="int", default=20, min=2, max=500),25            ParamSpec(name="ema_slow", type="int", default=80, min=5, max=2000),26            ParamSpec(name="vol_lookback", type="int", default=48, min=5, max=1000),27            ParamSpec(name="target_vol", type="float", default=0.40, min=0.01, max=3.0, description="Annualized vol target"),28            ParamSpec(name="max_leverage", type="float", default=1.0, min=0.0, max=1.0, description="Max exposure (spot → ≤1)"),29        ],30        failure_modes=["Whipsaw in range-bound regimes", "Fee drag at high turnover on short intervals", "Vol targeting lags regime shifts", "Only long leg available on spot"],31        references=["Moskowitz, Ooi, Pedersen (2012) Time series momentum, JFE"],32    )3334    def features(self, df: pd.DataFrame) -> pd.DataFrame:35        p = self.params36        f = pd.DataFrame(index=df.index)37        f["ts"] = df["ts"].values38        f["close"] = df["close"].values39        f["ret"] = log_returns(df["close"])40        f["mom"] = np.log(df["close"] / df["close"].shift(p["lookback"]))41        f["ema_fast"] = df["close"].ewm(span=p["ema_fast"], adjust=False, min_periods=p["ema_fast"]).mean()42        f["ema_slow"] = df["close"].ewm(span=p["ema_slow"], adjust=False, min_periods=p["ema_slow"]).mean()43        bars_per_year = 365 * 24 * 60 / float(df.attrs.get("interval_min", 60))44        f["vol_scale"] = vol_target_scale(f["ret"], p["vol_lookback"], p["target_vol"], bars_per_year, cap=p["max_leverage"])45        return f4647    def target(self, feats: pd.DataFrame) -> pd.Series:48        signal = ((feats["mom"] > 0) & (feats["ema_fast"] > feats["ema_slow"])).astype(float)49        return (signal * feats["vol_scale"]).fillna(0.0)5051    def uncertainty(self, feats: pd.DataFrame) -> pd.Series | None:52        # |momentum| in units of rolling vol → weak signals near zero are the uncertain ones53        z = feats["mom"] / (feats["ret"].rolling(self.params["lookback"], min_periods=5).std(ddof=1) * np.sqrt(self.params["lookback"]))54        return (1.0 / (1.0 + z.abs())).fillna(1.0)555657class MovingAverageEnsemble(Strategy):58    meta = StrategyMeta(59        name="ma_ensemble",60        label="Moving-average crossover ensemble",61        family="trend",62        hypothesis="An ensemble of EMA crossover rules at several horizons (vote fraction) reduces the parameter-fragility of a single "63        "crossover while keeping trend exposure. Exposure = fraction of rules that are bullish, volatility-scaled.",64        experimental=True,65        min_bars=200,66        params=[67            ParamSpec(name="fast_spans", type="choice", default="8,16,32", choices=["4,8,16", "8,16,32", "10,20,40"]),68            ParamSpec(name="slow_mult", type="int", default=3, min=2, max=10),69            ParamSpec(name="vol_lookback", type="int", default=48, min=5, max=1000),70            ParamSpec(name="target_vol", type="float", default=0.40, min=0.01, max=3.0),71        ],72        failure_modes=["Lag at turning points", "Choppy markets"],73    )7475    def features(self, df: pd.DataFrame) -> pd.DataFrame:76        p = self.params77        f = pd.DataFrame(index=df.index)78        f["ts"] = df["ts"].values79        f["close"] = df["close"].values80        f["ret"] = log_returns(df["close"])81        votes = []82        for s in [int(x) for x in p["fast_spans"].split(",")]:83            fast = df["close"].ewm(span=s, adjust=False, min_periods=s).mean()84            slow = df["close"].ewm(span=s * p["slow_mult"], adjust=False, min_periods=s * p["slow_mult"]).mean()85            col = f"x_{s}"86            f[col] = (fast > slow).astype(float).where(slow.notna())87            votes.append(col)88        f["vote"] = f[votes].mean(axis=1)89        bars_per_year = 365 * 24 * 60 / float(df.attrs.get("interval_min", 60))90        f["vol_scale"] = vol_target_scale(f["ret"], p["vol_lookback"], p["target_vol"], bars_per_year)91        return f9293    def target(self, feats: pd.DataFrame) -> pd.Series:94        return (feats["vote"] * feats["vol_scale"]).fillna(0.0)95