Warrior_EA/research/test_exits.py

"""Does trade management create edge, or only reshape it?

The user's point: "where we exit, where we put our stop losses matters as much as where
we enter." This tests it directly, and with the control that makes the answer meaningful.

Design: hold the ENTRIES fixed, vary only the EXIT rule, and measure expectancy in R.
Then run every identical exit rule on RANDOM entries at the same bars. If a management
rule creates expectancy, it will create it on random entries too - which would prove the
gain is not edge but a reshaping of the outcome distribution. Only a rule that beats its
own random-entry control by more than noise is doing something informational.

Exit rules swept:
  fixed        SL/TP at entry, the baseline
  be@1R        move stop to entry once +1R is touched
  trail_k      chandelier: stop trails k*ATR below the running high (long)
  time_T       flat at T bars regardless
  partial      half off at +1R, remainder runs to TP with stop at entry
"""
import numpy as np, sys, time
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
from classic import module_votes, combined_direction
from test_classic import prepare, sequential, barrier_outcomes, TF


def walk(o, h, l, a_sig, ent, dirs, sl_m, tp_m, H, sp, rule, param=0.0):
    """Bar-by-bar path simulation. Returns realised R per trade (R = the initial SL distance).

    Within-bar ordering is pessimistic throughout: the adverse barrier is tested before
    the favourable one, so a bar that spans both books the loss.
    """
    n = len(o)
    out = np.zeros(len(ent))
    for q, (e, d) in enumerate(zip(ent, dirs)):
        A = a_sig[e]
        if not np.isfinite(A) or A <= 0:
            continue
        R = sl_m * A
        entry = o[e] + (sp if d > 0 else -sp)
        stop = entry - d * R
        targ = entry + d * tp_m * A
        best = entry
        half_done = False
        pnl = 0.0
        closed = False
        for b in range(e, min(e + H, n)):
            hi, lo = h[b], l[b]
            # --- adverse first
            if (d > 0 and lo <= stop) or (d < 0 and hi >= stop):
                pnl += (1.0 if half_done else 1.0) * d * (stop - entry) / R
                closed = True
                break
            # --- favourable
            if (d > 0 and hi >= targ) or (d < 0 and lo <= targ):
                pnl += (0.5 if half_done else 1.0) * d * (targ - entry) / R
                closed = True
                break
            # --- management, applied on the CLOSE of the bar (never intrabar)
            best = max(best, hi) if d > 0 else min(best, lo)
            if rule == 'be@1R' and d * (best - entry) >= R:
                stop = max(stop, entry) if d > 0 else min(stop, entry)
            elif rule == 'trail':
                cand = best - d * param * A
                stop = max(stop, cand) if d > 0 else min(stop, cand)
            elif rule == 'partial':
                if not half_done and d * (best - entry) >= R:
                    pnl += 0.5 * 1.0          # half booked at +1R
                    half_done = True
                    stop = max(stop, entry) if d > 0 else min(stop, entry)
            elif rule == 'time' and (b - e) >= param:
                pnl += (0.5 if half_done else 1.0) * d * (c_close(o, b) - entry) / R
                closed = True
                break
        if not closed:
            b = min(e + H, n) - 1
            pnl += (0.5 if half_done else 1.0) * d * (c_close(o, b) - entry) / R
        out[q] = pnl
    return out


_CLOSE = {}


def c_close(o, b):
    return _CLOSE['c'][b]


def run(sym, tf, sl_m, tp_m, H, T=30, nrand=20, seed=7):
    o, h, l, c, a, sp, fireL, fireS, names, weights = prepare(sym, tf)
    _CLOSE['c'] = c
    n = len(c)
    a_sig = np.concatenate([[a[0]], a[:-1]])
    winL, resL, winS, resS = barrier_outcomes(o, h, l, a_sig, sl_m, tp_m, H, sp)

    votes = module_votes(fireL, fireS, weights)
    direction = combined_direction(votes)
    fb = np.nonzero(np.abs(direction) >= T)[0]
    d0 = np.sign(direction[fb]).astype(int)
    ent, dirs, _ = sequential(fb, d0, winL, winS, H, n)
    if len(ent) < 100:
        print(f"{sym} {sl_m}:{tp_m}  too few trades ({len(ent)})")
        return
    rng = np.random.default_rng(seed)

    rules = [('fixed', 0.0), ('be@1R', 0.0), ('trail 2.0', 2.0), ('trail 3.0', 3.0),
             ('trail 4.0', 4.0), ('partial', 0.0),
             ('time %d' % (H // 4), H // 4), ('time %d' % (H // 2), H // 2)]
    print(f"\n=== {sym} {TF.get(tf,tf)}  SL{sl_m}:TP{tp_m}  H={H}  vote>={T}  "
          f"{len(ent)} trades  (R = {sl_m} ATR) ===")
    print(f"{'exit rule':<12}{'signal E[R]':>13}{'random E[R]':>13}"
          f"{'diff':>9}{'diff sigma':>12}")
    for name, param in rules:
        base = name.split()[0]
        rr = walk(o, h, l, a_sig, ent, dirs, sl_m, tp_m, H, sp, base, param)
        sig_e = rr.mean()
        # control: identical bars and identical exit rule, random direction
        ctrl = np.empty(nrand)
        for k in range(nrand):
            rd = np.where(rng.random(len(ent)) < 0.5, 1, -1)
            ctrl[k] = walk(o, h, l, a_sig, ent, rd, sl_m, tp_m, H, sp, base, param).mean()
        diff = sig_e - ctrl.mean()
        sd = np.sqrt(rr.std(ddof=1) ** 2 / len(rr) + ctrl.var(ddof=1))
        print(f"{name:<12}{sig_e:>+13.4f}{ctrl.mean():>+13.4f}{diff:>+9.4f}"
              f"{diff/sd:>+12.2f}")


if __name__ == '__main__':
    t0 = time.time()
    for sym in ('EURUSD', 'USDJPY'):
        for (s, p, H) in [(2, 3, 96), (1, 2, 64)]:
            run(sym, 16385, s, p, H)
    print(f"\ntotal {time.time()-t0:.0f}s")