Warrior_EA/research/test_engine.py

"""Does the fill engine return ZERO when there is nothing there?

Every result in this project is a small number, and a harness with a small bias produces
small numbers indistinguishable from findings. Four results were retracted on 2026-08-02
because the harness had a bias nobody had measured. So before the engine is used again it
has to pass the tests that would have caught all four.

ARM 1 - REFLECTION IDENTITY (exact, per trade, no statistics)
-------------------------------------------------------------
Reflect the whole book about a constant: bid'(x) = 2C - ask(x), ask'(x) = 2C - bid(x).
This maps highs to lows, preserves the spread exactly, and turns every short into a long.
So for a correct engine, running a trade SHORT on the real book and the same trade LONG on
the reflected book must give the SAME R - not on average, but trade by trade, to floating
point. Drift cancels by construction because reflection flips the drift too, which is what
makes this stronger than any statistical symmetry check.

This is the test that catches a stop checked against the wrong side of the book, a gap
filled at the wrong price, or a tie broken inconsistently between longs and shorts.

ARM 2 - COST NULL (statistical)
-------------------------------
Random entries, symmetric barriers. A driftless market pays the round trip and nothing else:

        expR  ==  -(spread at the fill minute / risk)      within its standard error

Widening the stop divides the cost by k. If the residual does not shrink with it, the
residual is the harness rather than the spread.

The long-minus-short gap is REPORTED but not flagged: SP500 and gold rose for the whole
sample, so random longs and random shorts are genuinely different bets. Arm 1 is what
decides whether that gap is the market or the engine, and it does so without needing to
know the drift.
"""
import numpy as np, sys
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
import fills, book

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


class Mirror:
    """The book reflected about a constant. Quacks like fills.Book."""

    def __init__(self, bk, C):
        self.C = C; self.n = bk.n; self.t = bk.t
        self.bo, self.bh, self.bl, self.bc = 2 * C - bk.ao, 2 * C - bk.al, 2 * C - bk.ah, 2 * C - bk.ac
        self.ao, self.ah, self.al, self.ac = 2 * C - bk.bo, 2 * C - bk.bh, 2 * C - bk.bl, 2 * C - bk.bc

    def index_at(self, t_ms):
        return np.searchsorted(self.t, np.asarray(t_ms, np.int64), 'left')


def sample(sym, tf, k, n_trades, horizon_bars, seed, bk=None, f=None):
    bk = bk or fills.Book(sym)
    f = f or book.frame(sym, tf, bk)
    atr = f.atr(14)
    rng = np.random.default_rng(seed)
    e = np.unique(rng.integers(300, f.n - horizon_bars - 5, n_trades))
    e = e[np.isfinite(atr[e]) & (atr[e] > 0)]
    return bk, f, e, f.i0[e], f.c[e - 1], k * atr[e]


def reflection(sym, tf='H1', k=1.0, n_trades=8000, horizon_bars=100, seed=1):
    """Short on the real book vs the identical long on the reflected book."""
    bk, f, e, start, ref, risk = sample(sym, tf, k, n_trades, horizon_bars, seed)
    step = book.TF_SEC[tf] // 60
    H = horizon_bars * step
    m = len(e)
    short = fills.simulate(bk, start, -np.ones(m, int), ref + risk, ref - risk, H)
    C = float(np.median(f.c))
    mb = Mirror(bk, C)
    #--- the reflected levels: a stop ABOVE at ref+risk becomes a stop BELOW at 2C-(ref+risk)
    long_ = fills.simulate(mb, start, np.ones(m, int), 2 * C - (ref + risk),
                           2 * C - (ref - risk), H)
    a, b = short['R'], long_['R']
    if len(a) != len(b):
        return None, len(a), len(b), np.inf
    d = np.abs(a - b)
    return d.max(), len(a), int((d > 1e-9).sum()), float(np.abs(short['fill_px']
                                                               - (2 * C - long_['fill_px'])).max())


def cost_null(sym, tf='H1', k=1.0, n_trades=20000, horizon_bars=100, seed=0):
    bk, f, e, start, ref, risk = sample(sym, tf, k, n_trades, horizon_bars, seed)
    rng = np.random.default_rng(seed + 77)
    side = np.where(rng.random(len(e)) < 0.5, 1, -1)
    step = book.TF_SEC[tf] // 60
    out = fills.simulate(bk, start, side, ref - side * risk, ref + side * risk,
                         horizon_bars * step)
    if out is None:
        return None
    sp = (bk.ac - bk.bc)[out['idx']]
    return out, -(sp / out['risk']).mean()


if __name__ == '__main__':
    syms = [s for s in sys.argv[1:] if s in SYMS] or list(SYMS)

    print("=== ARM 1: REFLECTION IDENTITY ===")
    print("    a short on the real book must equal the same long on the reflected book,")
    print("    trade by trade. Any mismatch at all is an engine side bug.\n")
    print(f"  {'sym':>7}{'tf':>5}{'k':>5}{'n':>7}{'mismatched':>12}{'max |dR|':>12}{'max |dpx|':>12}")
    bad1 = 0
    for sym in syms:
        for tf in ('M15', 'H1'):
            for k in (1.0, 3.0):
                mx, n, nb, dpx = reflection(sym, tf, k=k)
                bad1 += bool(nb)
                print(f"  {sym:>7}{tf:>5}{k:>5.1f}{n:>7}{nb:>12}{mx:>12.2e}{dpx:>12.2e}"
                      + ('  <-- BUG' if nb else ''))
    print(f"  -> {'EXACT on every trade' if not bad1 else f'{bad1} cell(s) MISMATCH'}\n")

    print("=== ARM 2: COST NULL - random entries, symmetric barriers ===")
    print("    expR must equal the spread it paid. 'L-S' is reported for information: on a")
    print("    rising instrument random longs and shorts are different bets, and arm 1 has")
    print("    already ruled out the engine as the cause.\n")
    print(f"  {'sym':>7}{'tf':>5}{'k':>5}{'n':>7}"
          f"{'expR':>9}{'cost pred':>11}{'resid/se':>10}{'L-S':>9}{'same-bar':>10}{'unres':>8}")
    bad2 = 0
    for sym in syms:
        for tf in ('M15', 'H1'):
            for k in (1.0, 3.0):
                r = cost_null(sym, tf, k=k)
                if r is None:
                    continue
                out, pred = r
                R, sd = out['R'], out['side']
                se = R.std(ddof=1) / np.sqrt(len(R))
                resid = (R.mean() - pred) / max(se, 1e-12)
                ls = R[sd > 0].mean() - R[sd < 0].mean()
                flag = '  <-- CHECK' if abs(resid) > 3 else ''
                bad2 += bool(flag)
                print(f"  {sym:>7}{tf:>5}{k:>5.1f}{len(R):>7}"
                      f"{R.mean():>+9.4f}{pred:>+11.4f}{resid:>+10.2f}{ls:>+9.4f}"
                      f"{100*out['ambiguous']:>9.1f}%{100*out['unresolved']:>7.1f}%{flag}")
    print(f"  -> {'returns cost and nothing else' if not bad2 else f'{bad2} cell(s) off'}")