Warrior_EA/research/test_confluence.py

"""Confluence, not lone patterns.

The user's framing, and the EA's own design: weight-10 models are CONFIRMING states that
must never be acted on alone; a trade needs an event model plus agreement. So this tests
the aggregate, three ways:

  vote>=T   CExpertSignalCustom::Direction() - each module casts its highest-numbered
            matching pattern's weight, signed; the non-zero module votes are AVERAGED;
            trade when |average| >= T. This is literally what the shipped EA does.
  quorum>=K at least K of the 4 modules agree on a direction (any strength).
  event+cfm an ACTIONABLE model (weight > 10) fires, AND at least K confirming models
            (weight == 10) agree with it - the doctrine the weights encode.

Same discipline as test_classic.py: sequential non-overlapping trades, break-even ==
chance by the gambler's-ruin identity, family-wise 5% bar from a sign-flip null.
"""
import numpy as np, sys, time
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
from classic import module_votes, combined_direction, MODULE_SLICES
from test_classic import prepare, barrier_outcomes, sequential, TF


def evaluate(label, fire_bars, dirs, winL, winS, H, n, be, rng, nperm, acc):
    ti, td, tw = sequential(fire_bars, dirs, winL, winS, H, n)
    nT = len(ti)
    if nT < 30:
        return None
    wr = tw.mean()
    se = np.sqrt(be * (1 - be) / nT)
    z = (wr - be) / se
    wl_at, ws_at = winL[ti], winS[ti]
    fl = rng.random((nperm, nT)) < 0.5
    pz = (np.where(fl, wl_at[None, :], ws_at[None, :]).mean(axis=1) - be) / se
    acc.append(np.abs(pz))
    return (label, len(fire_bars), nT, 100 * wr, 100 * (wr - be), z)


def run(sym, tf, sl_m, tp_m, H, nperm=2000, seed=1):
    o, h, l, c, a, sp, fireL, fireS, names, weights = prepare(sym, tf)
    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)
    be = sl_m / (sl_m + tp_m)
    rng = np.random.default_rng(seed)
    rows, acc = [], []

    votes = module_votes(fireL, fireS, weights)
    direction = combined_direction(votes)
    V = np.column_stack([votes[m] for m in MODULE_SLICES])

    # --- 1. the EA's own averaged weighted vote
    for T in (10, 20, 30, 40, 50, 60, 70):
        fb = np.nonzero(np.abs(direction) >= T)[0]
        if len(fb) == 0:
            continue
        r = evaluate(f"vote>={T}", fb, np.sign(direction[fb]).astype(int),
                     winL, winS, H, n, be, rng, nperm, acc)
        if r:
            rows.append(r)

    # --- 2. quorum: K of 4 modules agree, no strength requirement
    agree_long = (V > 0).sum(axis=1)
    agree_short = (V < 0).sum(axis=1)
    for K in (2, 3, 4):
        m = ((agree_long >= K) & (agree_short == 0)) | ((agree_short >= K) & (agree_long == 0))
        fb = np.nonzero(m)[0]
        if len(fb) == 0:
            continue
        d = np.where(agree_long[fb] >= K, 1, -1)
        r = evaluate(f"quorum>={K}", fb, d, winL, winS, H, n, be, rng, nperm, acc)
        if r:
            rows.append(r)

    # --- 3. an actionable event, corroborated by K confirming states
    act = np.array([w > 10 for w in weights])
    cfm = ~act
    evL, evS = fireL[:, act].any(axis=1), fireS[:, act].any(axis=1)
    ncL, ncS = fireL[:, cfm].sum(axis=1), fireS[:, cfm].sum(axis=1)
    for K in (1, 2, 3, 4, 5):
        mL = evL & (ncL >= K) & ~evS
        mS = evS & (ncS >= K) & ~evL
        fb = np.nonzero(mL | mS)[0]
        if len(fb) == 0:
            continue
        d = np.where(mL[fb], 1, -1)
        r = evaluate(f"event+{K}cfm", fb, d, winL, winS, H, n, be, rng, nperm, acc)
        if r:
            rows.append(r)

    if not rows:
        return
    crit = np.quantile(np.maximum.reduce(acc), 0.95)
    print(f"\n=== {sym} {TF.get(tf,tf)}  SL{sl_m}:TP{tp_m}  H={H}  "
          f"break-even={100*be:.2f}%  (family-wise 5% bar |z|>{crit:.2f}) ===")
    print(f"{'rule':<14}{'fires':>9}{'trades':>8}{'win%':>8}{'edge pp':>9}{'z':>7}")
    for r in sorted(rows, key=lambda x: -x[5]):
        star = ' *' if abs(r[5]) > crit else ''
        print(f"{r[0]:<14}{r[1]:>9}{r[2]:>8}{r[3]:>8.2f}{r[4]:>+9.2f}{r[5]:>+7.2f}{star}")


if __name__ == '__main__':
    t0 = time.time()
    for sym in ('EURUSD', 'USDJPY', 'XAUUSD', 'SP500'):
        for (s, p, H) in [(2, 3, 96), (2, 6, 192), (1, 2, 64)]:
            try:
                run(sym, 16385, s, p, H)
            except Exception as ex:
                print(f"{sym} {s}:{p} FAILED {ex}")
    print(f"\ntotal {time.time()-t0:.0f}s")