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

"""MRV10 — Stochastic Reversion in Range (ADX-gated)

IDEA: Stochastic(14,3) oversold (<20) → long entry. Only trade when ADX<25 (range/sideways
regime, not a trending market). Exit when Stochastic crosses back above 50, or TP/SL hit.

This is a DISCRETE signal strategy (study_signals, 1d only).

Stochastic %K = 100 * (C - L_n) / (H_n - L_n)   [n=14 default]
Stochastic %D = SMA(%K, 3)                         [smoothed signal line]
ADX = average directional index (non-directional trend strength)

Grid: 2 param sets (stoch_period, adx_period) x (oversold threshold, adx_threshold)
  - Config A: stoch=14, %D<20, ADX<25, hold max 10 bars
  - Config B: stoch=14, %D<25, ADX<20, hold max 8 bars (stricter ADX)
Total: 2 configs -> 2 backtests (each on BTC+ETH) = 4 runs total.
"""
import sys
sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/alt")
import altlib as al
import numpy as np
import pandas as pd


def stochastic_k(df: pd.DataFrame, period: int = 14) -> np.ndarray:
    """Raw %K = (Close - LowestLow_n) / (HighestHigh_n - LowestLow_n) * 100. Causal."""
    hi = df["high"].values
    lo = df["low"].values
    c = df["close"].values
    n = len(c)
    k = np.full(n, np.nan)
    for i in range(period - 1, n):
        h_max = np.max(hi[i - period + 1: i + 1])
        l_min = np.min(lo[i - period + 1: i + 1])
        denom = h_max - l_min
        if denom > 0:
            k[i] = 100.0 * (c[i] - l_min) / denom
        else:
            k[i] = 50.0  # flat candle
    return k


def stochastic_d(k: np.ndarray, smooth: int = 3) -> np.ndarray:
    """Stochastic %D = SMA(%K, smooth). Causal."""
    return pd.Series(k).rolling(smooth, min_periods=smooth).mean().values


def adx(df: pd.DataFrame, period: int = 14) -> np.ndarray:
    """ADX (Average Directional Index). Causal, EMA-smoothed."""
    hi = df["high"].values
    lo = df["low"].values
    c = df["close"].values
    n = len(c)

    pc = np.roll(c, 1)
    pc[0] = c[0]
    ph = np.roll(hi, 1)
    ph[0] = hi[0]
    pl = np.roll(lo, 1)
    pl[0] = lo[0]

    tr = np.maximum(hi - lo, np.maximum(np.abs(hi - pc), np.abs(lo - pc)))
    dm_plus = np.where((hi - ph) > (pl - lo), np.maximum(hi - ph, 0.0), 0.0)
    dm_minus = np.where((pl - lo) > (hi - ph), np.maximum(pl - lo, 0.0), 0.0)

    # Wilder smoothing (like EMA with alpha=1/period)
    atr_s = pd.Series(tr).ewm(alpha=1.0 / period, adjust=False).mean().values
    dmp_s = pd.Series(dm_plus).ewm(alpha=1.0 / period, adjust=False).mean().values
    dmm_s = pd.Series(dm_minus).ewm(alpha=1.0 / period, adjust=False).mean().values

    di_plus = np.where(atr_s > 0, 100.0 * dmp_s / atr_s, 0.0)
    di_minus = np.where(atr_s > 0, 100.0 * dmm_s / atr_s, 0.0)

    di_sum = di_plus + di_minus
    dx = np.where(di_sum > 0, 100.0 * np.abs(di_plus - di_minus) / di_sum, 0.0)
    adx_arr = pd.Series(dx).ewm(alpha=1.0 / period, adjust=False).mean().values
    return adx_arr


def make_entries_fn(stoch_period=14, stoch_smooth=3, os_thresh=20, adx_thresh=25, max_bars=10):
    """Returns an entries_fn(df) -> list[dict|None] for study_signals.

    Signal: go long when:
      - Stochastic %D crosses below os_thresh (oversold) from above
      - ADX < adx_thresh (range regime, not trending)

    Exit: when %D crosses back above 50 OR max_bars elapsed.
    TP: 2 * ATR above entry. SL: 1.5 * ATR below entry.
    """
    def entries_fn(df: pd.DataFrame):
        k = stochastic_k(df, stoch_period)
        d = stochastic_d(k, stoch_smooth)
        adx_vals = adx(df, stoch_period)
        atr_vals = al.atr(df, stoch_period)
        c = df["close"].values
        n = len(df)

        entries = [None] * n
        for i in range(2, n):
            if np.isnan(d[i]) or np.isnan(d[i - 1]) or np.isnan(adx_vals[i]):
                continue
            # Oversold cross: %D was above threshold, now crossed below
            crossed_oversold = (d[i - 1] >= os_thresh) and (d[i] < os_thresh)
            in_range = adx_vals[i] < adx_thresh

            if crossed_oversold and in_range:
                atr_i = atr_vals[i] if np.isfinite(atr_vals[i]) and atr_vals[i] > 0 else c[i] * 0.01
                tp = c[i] + 2.0 * atr_i
                sl = c[i] - 1.5 * atr_i
                entries[i] = {
                    "dir": +1,
                    "tp": tp,
                    "sl": sl,
                    "max_bars": max_bars,
                }
        return entries

    return entries_fn


# ── Grid (small: 2 configs only) ──────────────────────────────────────────────
CONFIGS = [
    dict(label="CFG-A", stoch_period=14, stoch_smooth=3, os_thresh=20, adx_thresh=25, max_bars=10),
    dict(label="CFG-B", stoch_period=14, stoch_smooth=3, os_thresh=25, adx_thresh=20, max_bars=8),
]

if __name__ == "__main__":
    best_rep = None
    best_hold = -99.0

    for cfg in CONFIGS:
        label = cfg.pop("label")
        fn = make_entries_fn(**cfg)
        name = f"MRV10-{label}"
        print(f"\n--- Running {name} ---")
        rep = al.study_signals(name, fn, tfs=("1d",))
        print(al.fmt(rep))
        hold = rep["verdict"].get("best_holdout_sharpe", -99.0) or -99.0
        if hold > best_hold:
            best_hold = hold
            best_rep = rep
        cfg["label"] = label  # restore for logging

    print("\n\n=== BEST CONFIG ===")
    print(al.fmt(best_rep))
    print("JSON:", al.as_json(best_rep))