Warrior_EA/research/check_fill.py

"""Does the EURUSD fade survive a HONEST intrabar fill?

The mirror test entered at the retail trigger price e2 but started the outcome race at the
OPEN of the fill bar. Price at that open is on the far side of e2 - that is why the entry is
a stop order in the first place. So the race began before price had actually reached the
entry, handing the fade a free run toward its target and pushing the stop further away than
it really was.

That is a bias in the fade's favour and it is the same shape as the one that produced +0.7 R
in the swing test. It has to be measured, not argued about: find the first M5 bar that
actually trades AT e2, start the race there, and see what is left.
"""
import numpy as np, sys
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
from test_retail import setups, triggered, race_px, load_bars

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


def run(sym, tf, H=200, ms=(1.0,), path_tf='M5', honest=True):
    ev, o, h, l, c, spm, tick = setups(sym, tf)
    a1, I1 = load_bars(sym, tf)
    t1 = a1[:, I1['time']].astype(np.int64)
    a2, I2 = load_bars(sym, path_tf)
    ph, pl, pc = a2[:, I2['high']], a2[:, I2['low']], a2[:, I2['close']]
    pmap = np.searchsorted(a2[:, I2['time']], t1)
    step = 12 if tf == 'H1' else 3
    HH = H * step
    out = []
    for name, idx, d, ent, stp in ev:
        ok, fill = triggered(h, l, idx, d, ent)
        if ok.sum() < 100:
            continue
        i2, d2, e2, s2 = fill[ok], d[ok], ent[ok], stp[ok]
        risk0 = np.abs(e2 - s2)
        g = risk0 > 2 * spm[i2]
        i2, d2, e2, risk0 = (v[g] for v in (i2, d2, e2, risk0))
        pi = np.clip(pmap[i2], 0, len(ph) - 1)
        start = pi
        if honest:
            #--- first M5 bar INSIDE the fill bar that actually trades at the entry level
            start = np.full(len(pi), -1, np.int64)
            live = np.ones(len(pi), bool)
            for k in range(0, step + 2):
                j = pi + k
                m = live & (j < len(ph))
                if not m.any():
                    break
                hit = np.where(d2[m] > 0, ph[j[m]] >= e2[m], pl[j[m]] <= e2[m])
                w = np.nonzero(m)[0][hit]
                start[w] = j[w]; live[w] = False
            good = start >= 0
            i2, d2, e2, risk0, start = (v[good] for v in (i2, d2, e2, risk0, start))
        keep = start + HH < len(ph)
        i2, d2, e2, risk0, start = (v[keep] for v in (i2, d2, e2, risk0, start))
        if len(start) < 200:
            continue
        sp = spm[np.maximum(i2 - 1, 0)]
        dd = -d2                                    # the fade
        for m_ in ms:
            R0 = np.maximum(m_ * risk0, 2 * sp)
            r = race_px(ph, pl, start, dd, e2 - dd * R0, e2 + dd * R0, HH)
            R = np.where(r > 0, 1.0, np.where(r < 0, -1.0, 0.0))
            un = r == 0
            if un.any():
                q = np.minimum(start[un] + HH, len(pc) - 1)
                R[un] = (pc[q] - e2[un]) * dd[un] / R0[un]
            R = R - sp / R0
            se = R.std(ddof=1) / np.sqrt(len(R))
            f = np.array_split(R, 4)
            out.append((name, m_, len(R), R.mean(), R.mean() / max(se, 1e-12),
                        [float(x.mean()) for x in f]))
    return out


if __name__ == '__main__':
    syms = [s for s in sys.argv[1:] if s in SYMS] or ['EURUSD']
    for honest, lbl in ((False, 'race from the BAR OPEN (what the mirror test did)'),
                        (True, 'race from the ACTUAL FILL (honest)')):
        print(f"\n=== {lbl} ===")
        print(f"  {'symbol':>7}{'tf':>5} {'setup':<8}{'n':>7}{'fade expR':>11}{'t':>8}"
              f"   {'Q1':>7}{'Q2':>7}{'Q3':>7}{'Q4':>7}")
        for tf in ('M15', 'H1'):
            for s in syms:
                for r in run(s, tf, honest=honest):
                    print(f"  {s:>7}{tf:>5} {r[0]:<8}{r[2]:>7}{r[3]:>+11.4f}{r[4]:>+8.2f}"
                          + "".join(f"{x:>+7.3f}" for x in r[5]))