Warrior_EA/research/test_atrscale.py

"""Why does entering at a breakout do WORSE than entering at a random nearby bar?

The drift run found it in both directions at once: real minus local control is -0.045 R on
average and reaches -0.28 in a cell, for longs AND shorts. A directional edge cannot be
negative on both sides, so this is not about direction - it is about the MOMENT.

The suspect is barrier scaling. A breakout bar has an elevated ATR by construction. Stops and
targets are set at multiples of that ATR, so they are sized for a volatility that is, on
average, about to fall back. The trade then sits inside barriers that are too wide for the
market it is actually in.

This matters well beyond the research track: the EA sizes its own SL/TP from ATR presets, and
if ATR at a signal is systematically unrepresentative of the ATR that follows, every one of
those presets is mis-scaled in the same direction.

WHAT IS MEASURED
----------------
  ratio   ATR at entry / realised ATR over the holding window - >1 means the barriers were
          set from a volatility that did not persist
  unres   fraction that never touched either barrier - the direct symptom of barriers too
          wide for the market
  held    bars held
  win     fraction that reached the target

and then the fix that follows from the diagnosis: size the barriers from a SLOWER ATR
(a 100-bar average rather than a 14-bar one), which is not elevated at a breakout, and see
whether the real-minus-control gap closes.
"""
import numpy as np, sys
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
import fills, book, wyckoff

SYMS = ('EURUSD', 'USDJPY', 'XAUUSD', 'SP500')


def trade(bk, f, bars, d, risk_src, mrisk, kR, H, step):
    ok = (bars > 300) & (bars < f.n - 5) & np.isfinite(risk_src[bars]) & (risk_src[bars] > 0)
    bars, d = bars[ok], d[ok]
    start = f.i0[bars]
    ent = np.where(d > 0, bk.ao[start], bk.bo[start])
    risk = mrisk * risk_src[bars]
    o = fills.simulate(bk, start, d, ent - d * risk, ent + d * kR * risk, H * step,
                       entry=fills.MARKET)
    if o is None:
        return None
    o['indep'] = book.nonoverlap(o['idx'], o['exit_idx'] - o['idx'])
    o['bars'] = bars[np.nonzero(o['filled'])[0][o['kept']]]
    return o


def realised_atr(f, bars, H):
    """Mean true range over the H bars AFTER entry - the volatility the trade actually met."""
    tr = np.maximum(f.h - f.l, np.maximum(np.abs(f.h - np.roll(f.c, 1)),
                                          np.abs(f.l - np.roll(f.c, 1))))
    tr[0] = f.h[0] - f.l[0]
    cs = np.concatenate([[0.0], np.cumsum(tr)])
    j = np.minimum(bars + H, len(tr))
    return (cs[j] - cs[bars]) / np.maximum(j - bars, 1)


def describe(f, o, H, src):
    R = o['R'][o['indep']]
    b = o['bars'][o['indep']]
    ratio = src[b] / np.maximum(realised_atr(f, b, H), 1e-12)
    return (R.mean(), book.tstat(R), o['unresolved'], float(np.median(ratio)),
            float(np.mean(o['bars_held'][o['indep']])), float((R > 0.5).mean()), len(R))


def run(sym, tf, mrisk=3.0, kR=2.0, H=200, seed=3, span=250, slow=100):
    bk = fills.Book(sym)
    f = book.frame(sym, tf, bk)
    step = book.TF_SEC[tf] // 60
    ev = wyckoff.breakouts(f, score=False)
    if ev is None:
        return None
    fast = f.atr(14)
    #--- a SLOW ATR is not elevated by the breakout bar itself
    sl = f.atr(slow)
    rng = np.random.default_rng(seed)
    out = []
    for name, src in (('atr14', fast), (f'atr{slow}', sl)):
        for d0 in (+1, -1):
            m = ev['d'] == d0
            if m.sum() < 150:
                continue
            bars = ev['i'][m] + 1
            ctl = np.clip(bars + rng.integers(-span, span + 1, len(bars)), 301, f.n - 6)
            a = trade(bk, f, bars, np.full(m.sum(), d0), src, mrisk, kR, H, step)
            b = trade(bk, f, ctl, np.full(m.sum(), d0), src, mrisk, kR, H, step)
            if a is None or b is None:
                continue
            out.append((name, d0, describe(f, a, H, src), describe(f, b, H, src)))
    return out


if __name__ == '__main__':
    syms = [s for s in sys.argv[1:] if s in SYMS] or list(SYMS)
    print("=== WHY BREAKOUT ENTRIES UNDERPERFORM NEARBY RANDOM ONES ===")
    print("    'ratio' = ATR used for the barriers / ATR actually realised. >1 means the")
    print("    barriers were sized from a volatility that did not persist.\n")
    print(f"  {'sym':>7}{'tf':>4}{'sizedby':>9}{'side':>6}"
          f"{'REAL':>9}{'ctrl':>9}{'gap':>9}"
          f"{'ratio R':>9}{'ratio C':>9}{'unres R':>9}{'unres C':>9}{'win R':>8}{'win C':>8}")
    gaps = {}
    for sym in syms:
        for tf in ('H1',):
            r = run(sym, tf)
            if not r:
                continue
            for name, d0, A, B in r:
                gap = A[0] - B[0]
                gaps.setdefault(name, []).append(gap)
                print(f"  {sym:>7}{tf:>4}{name:>9}{'long' if d0>0 else 'short':>6}"
                      f"{A[0]:>+9.4f}{B[0]:>+9.4f}{gap:>+9.4f}"
                      f"{A[3]:>9.2f}{B[3]:>9.2f}{100*A[2]:>8.1f}%{100*B[2]:>8.1f}%"
                      f"{100*A[5]:>7.1f}%{100*B[5]:>7.1f}%")
    print()
    for name, g in gaps.items():
        g = np.array(g)
        print(f"  sized by {name:>7}: mean real-control {g.mean():+.4f}, "
              f"positive {int((g>0).sum())}/{len(g)}")
    print("\n  If the gap closes when the barriers are sized from the slow ATR, the effect")
    print("  was barrier mis-scaling at the signal bar, not the signal itself.")