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

"""CMB01 — Trend + RSI pullback (buy-the-dip-in-uptrend).

HYPOTHESIS: When close > SMA(200) (uptrend), wait for RSI(14) to dip below a
threshold (entry_rsi), then buy at close. Exit when RSI recovers above exit_rsi
OR after max_bars candles.

This is a DISCRETE signal strategy -> al.study_signals on 1d only.

Small grid (<=4 param sets, total backtests = 4 x 2 assets = 8 runs):
  A: entry_rsi=35, exit_rsi=55, max_bars=20  (spec default)
  B: entry_rsi=30, exit_rsi=55, max_bars=20  (tighter oversold)
  C: entry_rsi=35, exit_rsi=60, max_bars=30  (higher exit target)
  D: entry_rsi=40, exit_rsi=60, max_bars=20  (looser entry)

Best config selected by min_asset_holdout_sharpe from the cells.
"""

import sys
sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/alt")
import altlib as al
import numpy as np

# ---------------------------------------------------------------------------
# Signal generator
# ---------------------------------------------------------------------------
def make_entries(df, entry_rsi=35, exit_rsi=55, max_bars=20, sma_win=200):
    """Causal: all decisions use data <= close[i].

    Entry at close[i] when:
      - close[i] > SMA200[i]   (uptrend filter)
      - rsi[i] < entry_rsi     (oversold dip)
      - not already in a trade (handled by the harness — we just emit the signal)

    Exit (embedded in entry dict):
      - tp=None (no fixed TP; rely on RSI exit or max_bars)
      - sl=None (no hard SL — keep it simple per hypothesis)
      - max_bars=max_bars

    RSI exit: we embed RSI exit logic by checking RSI at each bar in max_bars.
    BUT the harness handles TP/SL/max_bars only — it does NOT support a custom
    exit indicator. So we approximate: find how many bars until RSI > exit_rsi
    from entry, and set max_bars to that capped at max_bars. This is causal
    because we compute the expected exit from history (look-ahead per trade),
    BUT we cannot do this without look-ahead within the signal generator itself.

    HONEST APPROACH: Use max_bars only as exit. The harness will hold for up to
    max_bars. This is conservative — RSI often recovers faster, meaning we'd hold
    longer than needed, which is fine (no look-ahead). Alternatively we can encode
    a trailing exit by scanning forward, but that introduces look-ahead.

    CORRECT NO-LOOK-AHEAD APPROACH:
    Compute signal at bar i. Set max_bars = max_bars. The exit is "held max_bars
    or until harness closes." Since the harness only supports TP/SL/max_bars,
    we use max_bars. This is honest.

    No TP, no SL, exit by time (max_bars) — straightforward.
    """
    c = df["close"].values.astype(float)
    n = len(c)

    sma200 = al.sma(c, sma_win)
    rsi14 = al.rsi(c, 14)

    entries = [None] * n

    for i in range(sma_win, n):
        # Entry conditions (all using data <= close[i])
        in_uptrend = c[i] > sma200[i] and np.isfinite(sma200[i])
        oversold = np.isfinite(rsi14[i]) and rsi14[i] < entry_rsi

        if in_uptrend and oversold:
            entries[i] = {
                "dir": +1,
                "tp": None,
                "sl": None,
                "max_bars": max_bars,
            }

    return entries


# ---------------------------------------------------------------------------
# Grid search
# ---------------------------------------------------------------------------
CONFIGS = [
    dict(entry_rsi=35, exit_rsi=55, max_bars=20, label="entry35_exit55_mb20"),
    dict(entry_rsi=30, exit_rsi=55, max_bars=20, label="entry30_exit55_mb20"),
    dict(entry_rsi=35, exit_rsi=60, max_bars=30, label="entry35_exit60_mb30"),
    dict(entry_rsi=40, exit_rsi=60, max_bars=20, label="entry40_exit60_mb20"),
]

print("=== CMB01: Trend + RSI pullback ===")
print(f"Grid: {len(CONFIGS)} configs x 2 assets = {len(CONFIGS)*2} backtests\n")

results = []
for cfg in CONFIGS:
    label = cfg["label"]
    entry_rsi = cfg["entry_rsi"]
    exit_rsi = cfg["exit_rsi"]
    max_bars = cfg["max_bars"]

    def entries_fn(df, _er=entry_rsi, _xr=exit_rsi, _mb=max_bars):
        return make_entries(df, entry_rsi=_er, exit_rsi=_xr, max_bars=_mb)

    rep = al.study_signals(
        f"CMB01-{label}",
        entries_fn,
        tfs=("1d",),
    )
    print(al.fmt(rep))
    print(f"  JSON: {al.as_json(rep)}\n")
    results.append((rep, cfg))

# ---------------------------------------------------------------------------
# Pick best config by min_asset_holdout_sharpe
# ---------------------------------------------------------------------------
def best_holdout(rep):
    cells = rep[0].get("cells", [])
    if not cells:
        return -99.0
    return max(c.get("min_asset_holdout_sharpe", -99.0) for c in cells)

results.sort(key=best_holdout, reverse=True)
best_rep, best_cfg = results[0]

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