Warrior_EA/research/test_swing_fade.py

"""Generalise the trigger: fade breakouts of SWING extremes, not named candlestick patterns.

The user's point, and it is the right one: we do not need to encode every retail pattern.
Retail buys the break of a swing high and puts stops under the swing low, and Wyckoff
labels the same locations. If what the pin/inside fade is really capturing is "a stop order
filled at a local extreme reverts", then the extreme is the ingredient and the candlestick
name is decoration.

This tests exactly that, and it is the honest way to find out whether the earlier result was
a pattern or a location:

  trigger   price trades through the most recent CONFIRMED swing high/low
  fade      take the other side at that level (a limit order into their stop/breakout buying)
  stop      m x ATR - deliberately decoupled from any pattern's geometry
  target    1 x risk

Confirmation lag is the thing to get right. A swing high at bar i is only known at bar i+N,
so the level may only be USED from i+N onward. Using it earlier is the classic fractal
lookahead and it would make any of this look wonderful.
"""
import numpy as np, sys
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
from test_retail import load_bars, race_px, PIP
from test_cause_effect import atr_of

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


def swings(h, l, N):
    """Confirmed swing highs/lows. Returned as arrays holding, for each bar, the most recent
    swing level that is ALREADY CONFIRMED at that bar (NaN until one exists)."""
    n = len(h)
    hi = np.full(n, np.nan); lo = np.full(n, np.nan)
    W = np.lib.stride_tricks.sliding_window_view
    if n < 2 * N + 2:
        return hi, lo
    wh = W(h, 2 * N + 1); wl = W(l, 2 * N + 1)
    is_hi = wh.argmax(axis=1) == N            # centre bar is the max of its window
    is_lo = wl.argmin(axis=1) == N
    cur_h = np.nan; cur_l = np.nan
    for i in range(n):
        #--- a pivot centred at c is confirmed at c+N; index into the window arrays
        c = i - N
        k = c - N
        if 0 <= k < len(is_hi):
            if is_hi[k]:
                cur_h = h[c]
            if is_lo[k]:
                cur_l = l[c]
        hi[i] = cur_h; lo[i] = cur_l
    return hi, lo


def run(sym, tf='H1', Ns=(3, 5, 8), ms=(0.25, 0.5, 1.0), H=200, path_tf='M5'):
    a1, I1 = load_bars(sym, tf)
    g = lambda k: a1[:, I1[k]]
    o, h, l, c = g('open'), g('high'), g('low'), g('close')
    spm = g('spread_mean')
    t1 = a1[:, I1['time']].astype(np.int64)
    atr = atr_of(h, l, c, 14); atr = np.concatenate([[atr[0]], atr[:-1]])
    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)
    HH = H * (12 if tf == 'H1' else 3)
    out = []
    for N in Ns:
        shi, slo = swings(h, l, N)
        for d, lvl in ((-1, shi), (+1, slo)):     # -1 = fade an upside break
            #--- break happens on bar i if the level was confirmed BEFORE i and price
            #--- trades through it during i
            prev = np.concatenate([[np.nan], lvl[:-1]])
            brk = (np.isfinite(prev) &
                   (h > prev if d < 0 else l < prev) &
                   (np.concatenate([[np.nan], h[:-1]]) <= prev if d < 0
                    else np.concatenate([[np.nan], l[:-1]]) >= prev))
            idx = np.nonzero(brk)[0]
            idx = idx[(idx > 2 * N + 20) & (idx < len(c) - 5)]
            if len(idx) < 300:
                continue
            entry = prev[idx]
            pi = np.clip(pmap[idx], 0, len(ph) - 1)
            keep = pi + HH < len(ph)
            idx, entry, pi = idx[keep], entry[keep], pi[keep]
            if len(idx) < 300:
                continue
            j = np.maximum(idx - 1, 0)
            sp = spm[j]
            dd = np.full(len(idx), d)
            for m in ms:
                R0 = np.maximum(m * atr[j], 2 * sp)
                r = race_px(ph, pl, pi, dd, entry - dd * R0, entry + 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(pi[un] + HH, len(pc) - 1)
                    R[un] = (pc[q] - entry[un]) * dd[un] / R0[un]
                R = R - sp / R0
                se = R.std(ddof=1) / np.sqrt(len(R))
                out.append((N, 'break^' if d < 0 else 'breakv', m, len(R),
                            (sp / R0).mean(), R.mean(), R.mean() / max(se, 1e-12)))
    return out


if __name__ == '__main__':
    syms = [s for s in sys.argv[1:] if s in SYMS] or list(SYMS)
    print("=== FADE THE BREAK OF A CONFIRMED SWING EXTREME ===")
    print("    the location, stripped of any candlestick name. stop = m x ATR.\n")
    print(f"  {'symbol':>7}{'tf':>5}{'N':>4}{'side':>8}{'m':>6}{'trades':>8}"
          f"{'cost(R)':>9}{'expR':>9}{'t':>8}")
    for tf in ('M15', 'H1'):
        for s in syms:
            for r in run(s, tf):
                print(f"  {s:>7}{tf:>5}{r[0]:>4}{r[1]:>8}{r[2]:>6.2f}{r[3]:>8}"
                      f"{r[4]:>9.3f}{r[5]:>+9.3f}{r[6]:>+8.2f}"
                      f"{'  <--' if r[5] > 0 and r[6] > 3 else ''}")