Warrior_EA/research/retrial.py

"""The four retracted positives, retried on an engine that has been proved correct.

On 2026-08-02 four results were withdrawn - a +0.53 R selection model, a +0.15 R sweep entry,
a +0.7 R swing fade and the headline +0.097 R EURUSD retail fade. All four shared one bug:
the trade entered at a LEVEL but the outcome clock started at the bar's OPEN, which sits on
the far side of that level by construction. They deserve a fair trial rather than a verdict
inherited from a broken harness, so each is rebuilt here on `book.py` + `fills.py`, where the
fill index IS the start index and the reflection identity holds exactly.

WHAT IS DIFFERENT THIS TIME, CONCRETELY
---------------------------------------
  - a buy stop triggers when the ASK reaches it, a sell limit when the BID does, and the race
    begins on that minute - not on the open of the bar that happened to contain it
  - the spread is the real spread on the fill minute, not a per-bar average
  - a gap past a stop fills at the open, which is where real slippage comes from
  - every headline is quoted on the NON-OVERLAPPING subset, because overlapping trades share
    price path and once turned t +1.61 into t +8.06

THE MIRROR TEST, PRESERVED
--------------------------
Retail's trade and its exact mirror under identical rules. Both pay the same spread and meet
the same tie convention, so those cancel in the difference and double in the sum:

    edge = (mirror - retail) / 2        cost = -(mirror + retail) / 2

The one change: retail enters on a STOP (they buy the break) and the mirror enters on a LIMIT
at the same price (the fade sells into that buying). Those are genuinely different orders and
fill at slightly different moments on a real book, which the old harness could not express.
"""
import numpy as np, sys
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
import fills, book
from fills import STOP, LIMIT, MARKET

SYMS = ('EURUSD', 'USDJPY', 'XAUUSD', 'SP500')
PIP = {'EURUSD': 1e-4, 'USDJPY': 1e-2, 'XAUUSD': 0.1, 'SP500': 0.1}


def sma(x, n):
    out = np.convolve(x, np.ones(n) / n, mode='full')[:len(x)]
    out[:n] = x[:n].mean()
    return out


def setups(f, tick, ma_n=20, slope_n=5, control=False, seed=5):
    """Retail's three mechanical setups, from the books, with their own stop rules.

    -> list of (name, fire_bar, direction, trigger_price, stop_price). Every level comes from
    bars at or before the fire bar; the order can only trigger later.

    `control=True` replaces the PATTERN with random bars drawn from the same trend-filter
    state, keeping the direction, the entry mechanics (a stop through the bar's extreme), the
    stop rule and the sample size. It answers the only question that matters once retail's
    own side looks positive: does the candlestick add anything over buying the break of ANY
    bar's high while the moving average is rising? On a drifting instrument the answer is
    usually no, and this is what shows it.
    """
    o, h, l, c = f.o, f.h, f.l, f.c
    m = sma(c, ma_n)
    up = np.zeros(len(c), bool); dn = np.zeros(len(c), bool)
    up[slope_n:] = (m[slope_n:] > m[:-slope_n]) & (c[slope_n:] > m[slope_n:])
    dn[slope_n:] = (m[slope_n:] < m[:-slope_n]) & (c[slope_n:] < m[slope_n:])
    rng = np.maximum(h - l, 1e-12)
    body = np.abs(c - o)
    upw = h - np.maximum(o, c); dnw = np.minimum(o, c) - l

    out = {}
    ins = np.zeros(len(c), bool)
    ins[1:] = (h[1:] < h[:-1]) & (l[1:] > l[:-1])
    out['inside'] = [(np.nonzero(ins & up)[0], +1, 'self'),
                     (np.nonzero(ins & dn)[0], -1, 'self')]
    bull_pin = (dnw >= 2 * body) & (dnw >= 0.5 * rng) & (c > (l + 0.5 * rng))
    bear_pin = (upw >= 2 * body) & (upw >= 0.5 * rng) & (c < (h - 0.5 * rng))
    out['pin'] = [(np.nonzero(bull_pin & up)[0], +1, 'prev'),
                  (np.nonzero(bear_pin & dn)[0], -1, 'prev')]
    be = np.zeros(len(c), bool); se = np.zeros(len(c), bool)
    be[1:] = (c[1:] > o[1:]) & (c[:-1] < o[:-1]) & (o[1:] <= c[:-1]) & (c[1:] >= o[:-1])
    se[1:] = (c[1:] < o[1:]) & (c[:-1] > o[:-1]) & (o[1:] >= c[:-1]) & (c[1:] <= o[:-1])
    out['engulf'] = [(np.nonzero(be & up)[0], +1, 'prev'),
                     (np.nonzero(se & dn)[0], -1, 'prev')]

    rng = np.random.default_rng(seed)
    pool = {+1: np.nonzero(up)[0], -1: np.nonzero(dn)[0]}
    ev = []
    for name, groups in out.items():
        for idx, d, stop_from in groups:
            idx = idx[(idx > ma_n + slope_n + 2) & (idx < len(c) - 400)]
            if not len(idx):
                continue
            if control:
                #--- same direction, same trend state, same count - pattern identity gone
                p = pool[d]
                p = p[(p > ma_n + slope_n + 2) & (p < len(c) - 400)]
                if len(p) < len(idx):
                    continue
                idx = np.sort(rng.choice(p, len(idx), replace=False))
            ent = np.where(d > 0, h[idx] + tick, l[idx] - tick)
            src = idx if stop_from == 'self' else idx - 1
            stp = np.where(d > 0, l[src] - tick, h[src] + tick)
            ev.append((name, idx, np.full(len(idx), d), ent, stp))
    return ev


def retail_arm(sym, tf='H1', k=1.0, H=200, control=False, bk=None, f=None):
    """Retail's OWN trade, entered honestly on a stop. -> per setup: R, independence mask."""
    bk = bk or fills.Book(sym)
    f = f or book.frame(sym, tf, bk)
    step = book.TF_SEC[tf] // 60
    tick = PIP[sym]
    res = []
    for name, idx, d, ent, stp in setups(f, tick, control=control):
        risk = np.abs(ent - stp)
        ok = risk > 4 * f.spread[idx]
        idx, d, ent, stp, risk = idx[ok], d[ok], ent[ok], stp[ok], risk[ok]
        if len(idx) < 200:
            continue
        start = f.i0[np.minimum(idx + 1, f.n - 1)]
        o = fills.simulate(bk, start, d, stp, ent + d * k * risk, H * step,
                           entry=STOP, entry_px=ent, entry_window=step)
        if o is None or o['n'] < 200:
            continue
        o['indep'] = book.nonoverlap(o['idx'], o['exit_idx'] - o['idx'])
        res.append((name, int(d[0]), o))
    return res


def mirror(sym, tf='H1', k=1.0, H=200, within=1, seed=0):
    """Retail's trade and its exact mirror, both entering at the SAME price level."""
    bk = fills.Book(sym)
    f = book.frame(sym, tf, bk)
    step = book.TF_SEC[tf] // 60
    tick = PIP[sym]
    rows = []
    for name, idx, d, ent, stp in setups(f, tick):
        risk = np.abs(ent - stp)
        ok = risk > 4 * f.spread[idx]                 # a stop inside the spread is not a trade
        idx, d, ent, stp, risk = idx[ok], d[ok], ent[ok], stp[ok], risk[ok]
        if len(idx) < 200:
            continue
        #--- the order is live from the NEXT bar, for `within` bars
        start = f.i0[np.minimum(idx + 1, f.n - 1)]
        win = within * step
        #--- retail: a stop order in the direction of the break
        ret = fills.simulate(bk, start, d, stp, ent + d * k * risk, H * step,
                             entry=STOP, entry_px=ent, entry_window=win)
        #--- the mirror: a limit order at the same price, the other way, stop reflected
        #--- about the entry so the two trades are geometrically identical
        mir = fills.simulate(bk, start, -d, ent + d * risk, ent - d * k * risk, H * step,
                             entry=LIMIT, entry_px=ent, entry_window=win)
        if ret is None or mir is None:
            continue
        rows.append((name, ret, mir, idx, d))
    return rows


def report(sym, tf, k, rows, span_bars, tf_step):
    for name, ret, mir, idx, d in rows:
        #--- both arms are restricted to the trades that BOTH filled, or the difference is
        #--- taken across two different populations and means nothing
        common = ret['filled'] & mir['filled']
        rk = common[ret['filled']]
        mk = common[mir['filled']]
        a = ret['R'][rk[ret['kept']]] if len(rk) == len(ret['kept']) else None
        yield name, ret, mir


def paired(sym, tf='H1', k=1.0, H=200):
    """expR of both arms on the trades that BOTH filled, plus edge/cost decomposition."""
    bk = fills.Book(sym)
    f = book.frame(sym, tf, bk)
    step = book.TF_SEC[tf] // 60
    tick = PIP[sym]
    res = []
    for name, idx, d, ent, stp in setups(f, tick):
        risk = np.abs(ent - stp)
        ok = risk > 4 * f.spread[idx]
        idx, d, ent, stp, risk = idx[ok], d[ok], ent[ok], stp[ok], risk[ok]
        if len(idx) < 200:
            continue
        start = f.i0[np.minimum(idx + 1, f.n - 1)]
        win = step
        arms = {}
        for tag, side, sl, tg, kind in (
                ('retail', d, stp, ent + d * k * risk, STOP),
                ('mirror', -d, ent + d * risk, ent - d * k * risk, LIMIT)):
            o = fills.simulate(bk, start, side, sl, tg, H * step,
                               entry=kind, entry_px=ent, entry_window=win)
            arms[tag] = o
        r, m = arms['retail'], arms['mirror']
        if r is None or m is None:
            continue
        #--- align on the events that produced a trade in BOTH arms
        def keymap(o):
            sel = np.nonzero(o['filled'])[0][o['kept']]
            return sel, o['R']
        sr, Rr = keymap(r); sm, Rm = keymap(m)
        common = np.intersect1d(sr, sm)
        if len(common) < 100:
            continue
        Rr = Rr[np.searchsorted(sr, common)]
        Rm = Rm[np.searchsorted(sm, common)]
        #--- non-overlapping on the event index, so no two trades share price path
        exr = r['exit_idx'][np.searchsorted(sr, common)]
        keep = book.nonoverlap(f.i0[np.minimum(idx[common] + 1, f.n - 1)],
                               exr - f.i0[np.minimum(idx[common] + 1, f.n - 1)])
        res.append(dict(name=name, n=len(common), Rr=Rr, Rm=Rm, keep=keep,
                        fill=r['fill_rate'], amb=r['ambiguous']))
    return res


if __name__ == '__main__':
    syms = [s for s in sys.argv[1:] if s in SYMS] or list(SYMS)
    print("=== RETRIAL 1: the retail fade, on the validated engine ===")
    print("    edge = (mirror - retail)/2 ; cost = -(mirror + retail)/2")
    print("    'indep' repeats the edge on the non-overlapping subset - the honest n.\n")
    print(f"  {'sym':>7}{'tf':>5}{'setup':>8}{'k':>4}{'n':>7}"
          f"{'retail':>9}{'mirror':>9}{'EDGE':>9}{'cost':>8}"
          f"{'t(edge)':>9}{'indep n':>9}{'indep':>9}{'t':>7}")
    for sym in syms:
        for tf in ('M15', 'H1'):
            for k in (1.0, 2.0):
                for r in paired(sym, tf, k=k):
                    Rr, Rm, keep = r['Rr'], r['Rm'], r['keep']
                    edge = 0.5 * (Rm - Rr)
                    cost = -0.5 * (Rm + Rr)
                    ei = edge[keep]
                    print(f"  {sym:>7}{tf:>5}{r['name']:>8}{k:>4.0f}{r['n']:>7}"
                          f"{Rr.mean():>+9.4f}{Rm.mean():>+9.4f}{edge.mean():>+9.4f}"
                          f"{cost.mean():>+8.4f}{book.tstat(edge):>+9.2f}"
                          f"{int(keep.sum()):>9}{ei.mean():>+9.4f}{book.tstat(ei):>+7.2f}")