spb/crypto-trade-bot

1"""Mean reversion: z-score of price vs rolling mean with OU half-life estimate and a trend-regime filter."""23from __future__ import annotations45import numpy as np6import pandas as pd78from ktrade.strategy.base import ParamSpec, Strategy, StrategyMeta, log_returns91011def ou_half_life(x: pd.Series) -> float:12    """Half-life from an AR(1) fit of Δx on x_{t-1}: HL = -ln2 / ln(1+β). Returns nan when not mean-reverting."""13    x = x.dropna()14    if len(x) < 20:15        return float("nan")16    dx = x.diff().dropna()17    xl = x.shift(1).dropna().loc[dx.index]18    xl_c = xl - xl.mean()19    denom = float((xl_c**2).sum())20    if denom == 0:21        return float("nan")22    beta = float((xl_c * (dx - dx.mean())).sum() / denom)23    if beta >= 0:24        return float("nan")25    return float(-np.log(2) / np.log(1 + beta))262728class ZScoreMeanReversion(Strategy):29    meta = StrategyMeta(30        name="zscore_mr",31        label="Z-score mean reversion (OU half-life, regime-filtered)",32        family="mean_reversion",33        hypothesis="Short-horizon deviations of price from its rolling mean revert when the series is locally stationary "34        "(OU half-life short and finite). Enter long when z < -entry (oversold) only if the longer trend is not strongly negative; exit at z > -exit. "35        "Spot-only: the short leg (z > +entry) is not traded. Entries are skipped when the expected reversion is smaller than round-trip costs.",36        experimental=True,37        min_bars=200,38        params=[39            ParamSpec(name="window", type="int", default=48, min=10, max=2000, description="Rolling mean/std window"),40            ParamSpec(name="entry_z", type="float", default=2.0, min=0.5, max=5.0),41            ParamSpec(name="exit_z", type="float", default=0.5, min=0.0, max=3.0),42            ParamSpec(name="max_half_life", type="float", default=48.0, min=1.0, max=5000.0, description="Skip when OU half-life (bars) exceeds this"),43            ParamSpec(name="hl_window", type="int", default=240, min=30, max=5000),44            ParamSpec(name="trend_filter", type="int", default=240, min=0, max=5000, description="EMA span; no entries when price < EMA (0 = off)"),45            ParamSpec(name="cost_bps", type="float", default=60.0, min=0.0, max=500.0, description="Round-trip cost threshold for entry"),46        ],47        failure_modes=["Trending breakdown (catching falling knives)", "Half-life estimate noisy on short windows", "Costs dominate on 1-min bars"],48        references=["Uhlenbeck & Ornstein (1930)", "Chan (2013) Algorithmic Trading, ch. 2"],49    )5051    def features(self, df: pd.DataFrame) -> pd.DataFrame:52        p = self.params53        f = pd.DataFrame(index=df.index)54        f["ts"] = df["ts"].values55        f["close"] = df["close"].values56        lc = np.log(df["close"])57        f["ret"] = log_returns(df["close"])58        m = lc.rolling(p["window"], min_periods=p["window"]).mean()59        s = lc.rolling(p["window"], min_periods=p["window"]).std(ddof=1)60        f["z"] = (lc - m) / s.replace(0, np.nan)61        f["half_life"] = lc.rolling(p["hl_window"], min_periods=p["hl_window"]).apply(ou_half_life, raw=False)62        f["expected_rev_bps"] = (-f["z"] * s * 1e4).clip(lower=0)63        if p["trend_filter"] > 0:64            ema = df["close"].ewm(span=p["trend_filter"], adjust=False, min_periods=p["trend_filter"]).mean()65            f["trend_ok"] = (df["close"] >= ema * 0.97).astype(float).where(ema.notna())66        else:67            f["trend_ok"] = 1.068        return f6970    def target(self, feats: pd.DataFrame) -> pd.Series:71        p = self.params72        z = feats["z"].values73        hl = feats["half_life"].values74        ok = feats["trend_ok"].values75        erv = feats["expected_rev_bps"].values76        pos = np.zeros(len(z))77        holding = 0.078        for i in range(len(z)):79            if np.isnan(z[i]):80                pos[i] = 0.081                holding = 0.082                continue83            if holding == 0.0:84                if z[i] < -p["entry_z"] and (np.isnan(hl[i]) is False and hl[i] <= p["max_half_life"]) and ok[i] == 1.0 and erv[i] >= p["cost_bps"]:85                    holding = 1.086            else:87                if z[i] > -p["exit_z"]:88                    holding = 0.089            pos[i] = holding90        return pd.Series(pos, index=feats.index)9192    def uncertainty(self, feats: pd.DataFrame) -> pd.Series | None:93        hl = feats["half_life"]94        return (hl / self.params["max_half_life"]).clip(0, 1).fillna(1.0)95