Warrior_EA/research/pivot_expectancy.py

"""Pivot-to-pivot swing distances vs spread, and the expectancy of actually trading them.

The user's claim (2026-08-14, with a chart): "the spread is nothing compared to these moves -
target 1:2 or 1:3 RR of swings". Both halves get measured here, on the validated M1 bid/ask
book (see book.py / fills.py), SP500 H1 mid bars for signals, real ask/bid for every fill.

1. SWING SIZE - ATR-scaled causal ZigZag (reversal threshold k*ATR14): pivot-to-pivot
   amplitude in points, in ATR, in spreads. This quantifies the user's picture, and the
   user is RIGHT: swings are enormous relative to the spread.

2. PERFECT-FORESIGHT EXPECTANCY - buy every low pivot at the ask, sell every high pivot at
   the bid. This is the expectancy the picture suggests. It is not tradeable, because a
   pivot is only DEFINED by what price does afterwards.

3. CAUSAL EXPECTANCY - the honest version of the same trade. A pivot exists, causally, the
   moment price has retraced k*ATR from the extreme (the ZigZag's own confirmation rule -
   the same event that draws the leg on a chart). Strategy: when a new leg is confirmed on
   a CLOSED H1 bar, enter WITH the new leg at the next M1 minute's real ask/bid; exit when
   the opposite pivot confirms, same fill discipline. Every leg is traded, long and short
   reported separately (a swing edge must be a directional asymmetry - mirror-test rule).
   GROSS expectancy (mid-to-mid, zero cost) is reported next to NET, so the question
   "is it the spread that kills it?" is answered by subtraction.

What the confirmation rule costs is not a defect of ZigZag - it is the definition of a
swing. Any swing-trading rule pays it in its own coin: you learn the pivot happened only
after the market has already left it.
"""
import numpy as np
import sys

sys.stdout.reconfigure(encoding="utf-8", errors="replace")
import book
import fills

SYM = "SP500"
WARMUP = 50


def zigzag(f, k):
    """Causal ATR ZigZag on H1 mid bars. Extremes from highs/lows (true swing size);
    confirmation when the CLOSE has retraced k*ATR14 from the extreme.
    Returns list of (pivot_i, confirm_i, dir_of_ended_leg_extreme) where dir=+1 means the
    pivot is a swing HIGH (the leg into it went up; the new leg goes down)."""
    a = f.atr(14)  # ATR ending at the previous bar - usable at bar i
    piv = []
    up = True
    ext = f.h[WARMUP]
    ext_i = WARMUP
    for i in range(WARMUP + 1, f.n):
        th = k * a[i]
        if not np.isfinite(th) or th <= 0:
            continue
        if up:
            if f.h[i] > ext:
                ext, ext_i = f.h[i], i
            elif ext - f.c[i] >= th:
                piv.append((ext_i, i, +1))
                up, ext, ext_i = False, f.l[i], i
        else:
            if f.l[i] < ext:
                ext, ext_i = f.l[i], i
            elif f.c[i] - ext >= th:
                piv.append((ext_i, i, -1))
                up, ext, ext_i = True, f.h[i], i
    return piv, a


def main():
    bk = fills.Book(SYM)
    f = book.frame(SYM, "H1", bk)
    sp = float(np.nanmean(f.spread))
    print(f"{SYM} H1: {f.n} bars {np.datetime64(int(f.t[0]), 's')}..{np.datetime64(int(f.t[-1]), 's')}"
          f" | mean spread {sp:.2f} pts")
    for k in (1.0, 1.5, 2.0, 3.0):
        piv, a = zigzag(f, k)
        if len(piv) < 30:
            continue
        pi = np.array([p[0] for p in piv])       # pivot bar
        ci = np.array([p[1] for p in piv])       # confirmation bar
        dr = np.array([p[2] for p in piv])       # +1 swing high
        px = np.where(dr > 0, f.h[pi], f.l[pi])  # pivot price
        # --- 1. swing size: consecutive pivot-to-pivot distance
        amp = np.abs(np.diff(px))
        amp_atr = amp / a[pi[1:]]
        dur = np.diff(pi)
        lag = ci - pi                            # bars from pivot to knowing about it
        print(f"\n== reversal {k:.1f} x ATR: {len(amp)} legs "
              f"({1000.0 * len(amp) / f.n:.1f}/1000 bars), median duration {np.median(dur):.0f} bars, "
              f"median confirm lag {np.median(lag):.0f} bars ==")
        print(f"  swing size: median {np.median(amp):.1f} pts = {np.median(amp_atr):.2f} ATR = "
              f"{np.median(amp) / sp:.0f} spreads | mean {amp.mean():.1f} pts = {amp.mean() / sp:.0f} spreads")
        # --- 2. perfect foresight: every leg, pivot price to pivot price, one spread each
        print(f"  perfect foresight (untradeable): {amp.mean() - sp:+.1f} pts/leg net "
              f"({(amp.mean() - sp) / sp:.0f} spreads)")
        # --- 3. causal: enter at first M1 after the confirming bar closes, with the new leg;
        #        exit at first M1 after the NEXT confirmation. Real ask/bid opens.
        side = -dr[:-1]                           # after a swing high confirms, new leg is DOWN
        e = f.i0[np.minimum(ci[:-1] + 1, f.n - 1)]
        x = f.i0[np.minimum(ci[1:] + 1, f.n - 1)]
        ok = x > e
        entry_mid = 0.5 * (bk.ao[e] + bk.bo[e])
        exit_mid = 0.5 * (bk.ao[x] + bk.bo[x])
        gross = side * (exit_mid - entry_mid)
        net = np.where(side > 0, bk.bo[x] - bk.ao[e], bk.bo[e] - bk.ao[x])
        for name, m in (("ALL", ok), ("LONG", ok & (side > 0)), ("SHORT", ok & (side < 0))):
            g, nn = gross[m], net[m]
            n = len(g)
            se = g.std() / np.sqrt(n)
            wr = 100.0 * (nn > 0).mean()
            print(f"  causal {name:<5} n={n:5d}: GROSS {g.mean():+7.2f} pts/leg (t={g.mean() / se:+.1f}) | "
                  f"NET {nn.mean():+7.2f} | win {wr:.0f}% | vs swing size {amp[m].mean():.0f} pts")


if __name__ == "__main__":
    main()