PythagorasGoal/scripts/research/alt/runs/MRV03.py

"""MRV03 — Z-score reversion trend-gated (discrete signals, 1d).

HYPOTHESIS: Fade |zscore(close,20)| > 2 toward mean ONLY when the long-horizon
trend (SMA200 slope) is flat. Skip entries in strong trends.

Logic:
- z = zscore(close, 20): deviation from 20-bar mean
- slope = (SMA200[i] - SMA200[i-slope_win]) / SMA200[i-slope_win]: recent slope of SMA200
- Gate: |slope| < flat_thresh  → trend is flat → allow mean-reversion
- Entry: if z > +2 → SHORT (price too high, expect reversion to mean)
         if z < -2 → LONG (price too low, expect reversion to mean)
- Exit: TP at SMA20 (mean reversion target), SL at 3*ATR14, max_bars=10

Grid: 2 param sets (zscore_win x flat_thresh):
  A: zscore_win=20, flat_thresh=0.005
  B: zscore_win=20, flat_thresh=0.010
"""
import sys
sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/alt")
import altlib as al
import numpy as np

# ── CONFIG GRID (keep total backtests ≤ 6: 2 params × 1 TF × 2 assets = 4 per config) ──
CONFIGS = [
    dict(label="A", zscore_win=20, slope_win=5,  flat_thresh=0.005, z_thresh=2.0, max_bars=10),
    dict(label="B", zscore_win=20, slope_win=5,  flat_thresh=0.010, z_thresh=2.0, max_bars=10),
]


def make_entries_fn(zscore_win: int, slope_win: int, flat_thresh: float,
                    z_thresh: float, max_bars: int):
    """Return an entries_fn(df) for study_signals."""
    sma200_win = 200

    def entries_fn(df):
        c = df["close"].values.astype(float)
        n = len(c)

        # Indicators (all causal: value at i uses data <=i)
        z = al.zscore(c, zscore_win)
        sma20 = al.sma(c, zscore_win)          # mean-reversion target = rolling mean
        sma200 = al.sma(c, sma200_win)
        atr14 = al.atr(df, 14)

        # SMA200 slope: fractional change over last slope_win bars
        sma200_prev = np.full(n, np.nan)
        sma200_prev[slope_win:] = sma200[:-slope_win]
        slope = np.where(
            (sma200_prev > 0) & np.isfinite(sma200_prev),
            (sma200 - sma200_prev) / sma200_prev,
            np.nan,
        )

        entries = [None] * n
        for i in range(sma200_win + slope_win, n):
            zi = z[i]
            si = slope[i]
            ci = c[i]
            atr_i = atr14[i]
            m20_i = sma20[i]

            # NaN guard
            if not (np.isfinite(zi) and np.isfinite(si) and np.isfinite(ci)
                    and np.isfinite(atr_i) and np.isfinite(m20_i)):
                continue

            # Gate: trend must be flat
            if abs(si) >= flat_thresh:
                continue

            # Signal
            if zi > z_thresh:
                # Price is stretched UP → SHORT toward mean
                entries[i] = {
                    "dir": -1,
                    "tp": m20_i,                    # mean reversion target
                    "sl": ci + 3.0 * atr_i,        # stop above
                    "max_bars": max_bars,
                }
            elif zi < -z_thresh:
                # Price is stretched DOWN → LONG toward mean
                entries[i] = {
                    "dir": +1,
                    "tp": m20_i,                    # mean reversion target
                    "sl": ci - 3.0 * atr_i,        # stop below
                    "max_bars": max_bars,
                }

        return entries

    return entries_fn


def run():
    results = []
    for cfg in CONFIGS:
        print(f"\n--- Config {cfg['label']}: zscore_win={cfg['zscore_win']}, "
              f"slope_win={cfg['slope_win']}, flat_thresh={cfg['flat_thresh']}, "
              f"z_thresh={cfg['z_thresh']}, max_bars={cfg['max_bars']} ---")
        entries_fn = make_entries_fn(
            zscore_win=cfg["zscore_win"],
            slope_win=cfg["slope_win"],
            flat_thresh=cfg["flat_thresh"],
            z_thresh=cfg["z_thresh"],
            max_bars=cfg["max_bars"],
        )
        rep = al.study_signals(
            f"MRV03-{cfg['label']}",
            entries_fn,
            tfs=("1d",),
        )
        print(al.fmt(rep))
        print("JSON:", al.as_json(rep))
        results.append((cfg, rep))

    # Pick best config by min_asset_holdout_sharpe
    best_cfg, best_rep = max(
        results,
        key=lambda x: x[1]["verdict"].get("best_holdout_sharpe", -99),
    )
    print(f"\n=== BEST CONFIG: {best_cfg['label']} ===")
    print(al.fmt(best_rep))
    print("JSON:", al.as_json(best_rep))
    return best_rep


if __name__ == "__main__":
    run()