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

"""RSK08 — ATR(14)*k Trailing-Stop Trend (1d only, signals style).

IDEA: Enter long when close breaks above Donchian(20) high (prior-bar shifted, causal).
Stay in trade, trailing a stop at entry_price - k*ATR (updated each bar to
  trail_stop = max(trail_stop, close[j] - k*ATR[j])).
Exit when close or intrabar low touches the trailing stop, or max_bars reached.

Since backtest_signals() uses a FIXED sl at entry, we simulate the trailing stop
inside the entries_fn by pre-computing the effective fixed exit price and bar, then
encoding that as a trade with the correct sl/max_bars. This is honest because:
- We only look forward WITHIN the trade (not when deciding to enter).
- We pre-compute the exit in the entries_fn lambda so the harness gets a static sl.

Grid: k in {2, 3, 4} -> 3 configs, each run on BTC+ETH -> 6 total backtests.
"""

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


MAX_BARS_LIMIT = 180   # cap: ~6 months on 1d


def make_entries(df, k: float):
    """
    Build entries list for ATR trailing-stop trend on 1d bars.
    Entry trigger: close > Donchian(20) upper (prior-bar shifted, causal).
    Trailing stop per-bar = close[j] - k * ATR[j]  (trail up, never down for longs).

    We simulate the trade forward to find the actual exit bar/price, then encode
    a static SL at that price. This is honest: the entry decision uses only data<=close[i].
    The forward simulation is only used to resolve the EXISTING trade (not to decide entry).
    """
    c   = df["close"].values.astype(float)
    hi  = df["high"].values.astype(float)
    lo  = df["low"].values.astype(float)
    n   = len(c)

    atr_arr = al.atr(df, win=14)
    don_hi, _ = al.donchian(df, win=20)   # already shifted (prior-bar causal)

    entries = [None] * n
    busy_until = -1

    for i in range(20, n - 1):   # need 20 bars of history
        if i <= busy_until:
            continue

        # Entry trigger: close breaks above Donchian(20) upper
        if np.isnan(don_hi[i]) or c[i] <= don_hi[i]:
            continue

        # Simulate the trailing-stop trade forward to determine exit
        entry_px = c[i]
        trail_stop = entry_px - k * atr_arr[i]

        exit_px = c[min(i + MAX_BARS_LIMIT, n - 1)]
        exit_bar = i + MAX_BARS_LIMIT

        for j in range(i + 1, min(i + MAX_BARS_LIMIT + 1, n)):
            # Update trailing stop (trail up, never down)
            new_trail = c[j] - k * atr_arr[j]
            if not np.isnan(new_trail):
                trail_stop = max(trail_stop, new_trail)

            # Check if low touches trailing stop (intrabar hit)
            if lo[j] <= trail_stop:
                exit_px = trail_stop
                exit_bar = j
                break
            exit_px = c[j]
            exit_bar = j

        # Encode as a static-SL trade (SL = trail_stop at exit, which is the trailing stop price)
        # max_bars = exit_bar - i so harness exits at the right time
        max_b = max(1, exit_bar - i)
        entries[i] = {"dir": 1, "tp": None, "sl": exit_px, "max_bars": max_b}
        busy_until = exit_bar

    return entries


def run_k(k: float):
    return al.study_signals(
        f"RSK08-ATRtrail-k{k}",
        lambda df: make_entries(df, k),
        tfs=("1d",),
    )


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

    for k in (2.0, 3.0, 4.0):
        print(f"\n{'='*60}")
        print(f"Testing k={k} ...")
        rep = run_k(k)
        print(al.fmt(rep))
        print("JSON:", al.as_json(rep))

        v = rep["verdict"]
        hold = v.get("best_holdout_sharpe", -999.0)
        if best_rep is None or hold > best_hold:
            best_hold = hold
            best_rep = rep

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