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

"""TRD05 — ADX-filtered EMA crossover.

Hypothesis: EMA(fast, slow) cross provides directional signal ONLY when ADX(14) > threshold
(trending regime). When ADX is below the threshold (chop), position goes flat.

Grid (<=4 param sets, total backtests = 4 params * 2 assets * 2 tfs = 16, but we limit to 2 TFs):
  (fast_ema, slow_ema, adx_period, adx_thresh)
  - (20, 100, 14, 25)  — canonical from hypothesis
  - (10, 50, 14, 25)   — faster cross
  - (20, 100, 14, 20)  — more lenient ADX gate
  - (5, 20, 14, 25)    — short-term cross with ADX filter

We run 4 configs but only 1 TF at a time to stay within 2-CPU budget.
Best config selected by min-asset holdout Sharpe across 2 TFs (1d, 12h).

ADX calculation (causal):
  +DM[i] = max(high[i]-high[i-1], 0) if > (low[i-1]-low[i]) else 0
  -DM[i] = max(low[i-1]-low[i], 0) if > (high[i]-high[i-1]) else 0
  TR[i]  = max(high[i]-low[i], |high[i]-close[i-1]|, |low[i]-close[i-1]|)
  Smooth over `period` with Wilder's EMA (alpha=1/period)
  +DI = 100 * smooth(+DM) / smooth(TR)
  -DI = 100 * smooth(-DM) / smooth(TR)
  DX  = 100 * |+DI - -DI| / (+DI + -DI)
  ADX = Wilder EMA(DX, period)
"""
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 _wilder_ema(x: np.ndarray, period: int) -> np.ndarray:
    """Wilder smoothing (EMA with alpha=1/period, adjust=False)."""
    alpha = 1.0 / period
    out = np.empty(len(x), dtype=float)
    out[0] = x[0]
    for i in range(1, len(x)):
        out[i] = out[i - 1] * (1.0 - alpha) + x[i] * alpha
    return out


def _adx(df: pd.DataFrame, period: int = 14) -> np.ndarray:
    """Compute causal ADX(period). Returns array len(df), NaN for first ~2*period bars."""
    h = df["high"].values.astype(float)
    l = df["low"].values.astype(float)
    c = df["close"].values.astype(float)
    n = len(h)

    # True Range
    pc = np.roll(c, 1)
    pc[0] = c[0]
    tr = np.maximum(h - l, np.maximum(np.abs(h - pc), np.abs(l - pc)))

    # Directional Movements
    up = h - np.roll(h, 1)
    dn = np.roll(l, 1) - l
    up[0] = 0.0
    dn[0] = 0.0
    pos_dm = np.where((up > dn) & (up > 0), up, 0.0)
    neg_dm = np.where((dn > up) & (dn > 0), dn, 0.0)

    # Wilder smooth
    str_ = _wilder_ema(tr, period)
    spdm = _wilder_ema(pos_dm, period)
    sndm = _wilder_ema(neg_dm, period)

    # DI lines
    pdi = 100.0 * np.where(str_ > 0, spdm / str_, 0.0)
    ndi = 100.0 * np.where(str_ > 0, sndm / str_, 0.0)

    # DX and ADX
    denom = pdi + ndi
    dx = np.where(denom > 0, 100.0 * np.abs(pdi - ndi) / denom, 0.0)
    adx = _wilder_ema(dx, period)

    # First 2*period bars are warm-up — NaN them
    adx[:2 * period] = np.nan
    return adx


def make_target(fast: int, slow: int, adx_period: int, adx_thresh: float,
                vol_target: bool = True):
    """Return a target_fn for study_weights that implements ADX-filtered EMA cross."""
    def target_fn(df: pd.DataFrame) -> np.ndarray:
        c = df["close"].values.astype(float)

        ema_fast = al.ema(c, fast)
        ema_slow = al.ema(c, slow)
        adx_vals = _adx(df, adx_period)

        # Signal: +1 if fast > slow (bullish trend), -1 if fast < slow (bearish)
        # Flat when ADX < threshold (choppy) or ADX is NaN (warmup)
        cross_signal = np.where(ema_fast > ema_slow, 1.0, -1.0)

        trending = np.where(
            np.isfinite(adx_vals) & (adx_vals > adx_thresh),
            1.0, 0.0
        )

        direction = cross_signal * trending

        # Long-flat only (like TP01, we don't short crypto)
        # Actually let's try L/S first since hypothesis doesn't restrict
        direction_lf = np.clip(direction, 0, 1)  # long-flat version

        if vol_target:
            return al.vol_target(direction_lf, df, target_vol=0.20, vol_win_days=30,
                                 leverage_cap=2.0)
        else:
            return direction_lf

    return target_fn


# --- Grid of configs ---------------------------------------------------------
CONFIGS = [
    dict(fast=20, slow=100, adx_period=14, adx_thresh=25),  # canonical
    dict(fast=10, slow=50,  adx_period=14, adx_thresh=25),  # faster cross
    dict(fast=20, slow=100, adx_period=14, adx_thresh=20),  # relaxed gate
    dict(fast=5,  slow=20,  adx_period=14, adx_thresh=25),  # short-term
]

# We test 2 timeframes: 1d and 12h (within 2-CPU budget constraint)
TFS = ("1d", "12h")

best_rep = None
best_score = -999.0

print("=== TRD05: ADX-filtered EMA crossover ===\n")

for cfg in CONFIGS:
    label = f"TRD05(ema{cfg['fast']}/{cfg['slow']},adx{cfg['adx_period']}>{cfg['adx_thresh']})"
    fn = make_target(**cfg)
    rep = al.study_weights(label, fn, tfs=TFS)
    print(al.fmt(rep))
    print()

    # Score = min holdout sharpe across cells
    score = rep["verdict"].get("best_holdout_sharpe", -999.0) or -999.0
    if score > best_score:
        best_score = score
        best_rep = rep
        best_cfg = cfg

print("\n" + "=" * 60)
print(f"BEST CONFIG: {best_cfg}")
print(al.fmt(best_rep))
print("JSON:", al.as_json(best_rep))