forked from animatedread/Warrior_EA
c85f6a3839
Faithful Python port of ADZigZag (stock MetaQuotes ZigZag 12/5/3, verbatim rebrand) including the incremental prev_calculated branch, so the indicator can be replayed bar by bar exactly as it draws live. SP500 H1, 74,599 bars, fills at real M1 ask/bid: - FINAL swings: 4,658 legs, mean 50.8 pts = 95 spreads. Perfect foresight +50.2 pts/leg. The user premise (swings dwarf spread) is fully confirmed. - LIVE: 81% of drawn newest-pivots later repaint away entirely (18,805 of 23,299). Holding the drawn direction at every bar close grosses +0.38 pts/trade (t=0.8, zero cost charged) out of the 50.8-pt average swing - 0.7% of the line the chart ends up showing. - LONG +1.66 gross / SHORT -0.90 = the index drift, nothing else; long net +1.17 pts / 13-bar hold = ~1.5 bp, under one night financing. The spread subtracts 0.49 pts of a swing that hands over 0.38: cost was never the obstacle - pivot knowledge is. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
228 lines
9.3 KiB
Python
228 lines
9.3 KiB
Python
"""True swings from the EA's actual ZigZag (stock MetaQuotes algorithm: Depth 12,
|
|
Deviation 5, Backstep 3 - ADZigZag.mq5 is a verbatim rebrand), measured two ways:
|
|
|
|
1. FINAL swings - the lines the indicator leaves on a chart after history settles.
|
|
Pivot-to-pivot distance in points / ATR / spreads. This is what the user's screenshot
|
|
shows, and what "the spread is nothing compared to these moves" refers to.
|
|
|
|
2. LIVE swings - the indicator replayed bar by bar through its own incremental
|
|
recalculation (the prev_calculated>0 branch: rewind to the 3rd-last extreme, rescan),
|
|
which is what a chart shows AT THE TIME. The tradeable policy: at every H1 close, hold
|
|
the direction implied by the newest drawn pivot (newest pivot is a LOW -> the current
|
|
leg points up -> long; a HIGH -> short). Position flips become trades, filled at the
|
|
next M1 minute's real ask/bid from the validated book. This uses the indicator exactly
|
|
as drawn, no waiting, no hindsight - and also counts how often the drawn pivot it acted
|
|
on later repaints away.
|
|
|
|
The EA itself refuses to read a pivot younger than 100 bars (m_swingConfirmationBars),
|
|
because the last leg repaints; a live trader cannot wait 100 bars, so the honest live
|
|
policy above is the generous one - it acts immediately.
|
|
"""
|
|
import numpy as np
|
|
import sys
|
|
|
|
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
|
import book
|
|
import fills
|
|
|
|
SYM = "SP500"
|
|
DEPTH, DEVIATION, BACKSTEP = 12, 5, 3
|
|
POINT = 0.01 # SP500 quotes carry 2 decimals on this broker (e.g. 7779.26)
|
|
INIT_BARS = 300 # full-history first pass, then bar-by-bar increments
|
|
RECALC_EXTREMES = 3 # ADZZExtRecalc
|
|
|
|
|
|
def lowest(a, depth, start):
|
|
lo = start
|
|
for i in range(start - 1, max(start - depth, -1), -1):
|
|
if a[i] < a[lo]:
|
|
lo = i
|
|
return lo
|
|
|
|
|
|
def highest(a, depth, start):
|
|
hi = start
|
|
for i in range(start - 1, max(start - depth, -1), -1):
|
|
if a[i] > a[hi]:
|
|
hi = i
|
|
return hi
|
|
|
|
|
|
class ZigZag:
|
|
"""Faithful port of ADZigZag.mq5 OnCalculate, callable one bar at a time."""
|
|
|
|
def __init__(self, high, low, n_total):
|
|
self.h, self.l = high, low
|
|
self.zz = np.zeros(n_total)
|
|
self.hm = np.zeros(n_total)
|
|
self.lm = np.zeros(n_total)
|
|
self.prev = 0
|
|
|
|
def calculate(self, rates_total):
|
|
h, l, zz, hm, lm = self.h, self.l, self.zz, self.hm, self.lm
|
|
extreme_search = 0 # ADZZ_Extremum
|
|
curlow = curhigh = 0.0
|
|
if self.prev == 0:
|
|
start = DEPTH
|
|
else:
|
|
i = rates_total - 1
|
|
extreme_counter = 0
|
|
while extreme_counter < RECALC_EXTREMES and i > rates_total - 100:
|
|
if zz[i] != 0.0:
|
|
extreme_counter += 1
|
|
i -= 1
|
|
i += 1
|
|
start = i
|
|
if lm[i] != 0.0:
|
|
curlow = lm[i]
|
|
extreme_search = 1 # Peak
|
|
else:
|
|
curhigh = hm[i]
|
|
extreme_search = -1 # Bottom
|
|
zz[start + 1:rates_total] = 0.0
|
|
lm[start + 1:rates_total] = 0.0
|
|
hm[start + 1:rates_total] = 0.0
|
|
# --- searching for high and low extremes
|
|
last_low = last_high = 0.0
|
|
for shift in range(start, rates_total):
|
|
val = l[lowest(l, DEPTH, shift)]
|
|
if val == last_low:
|
|
val = 0.0
|
|
else:
|
|
last_low = val
|
|
if (l[shift] - val) > DEVIATION * POINT:
|
|
val = 0.0
|
|
else:
|
|
for back in range(1, BACKSTEP + 1):
|
|
res = lm[shift - back]
|
|
if res != 0.0 and res > val:
|
|
lm[shift - back] = 0.0
|
|
lm[shift] = val if l[shift] == val else 0.0
|
|
val = h[highest(h, DEPTH, shift)]
|
|
if val == last_high:
|
|
val = 0.0
|
|
else:
|
|
last_high = val
|
|
if (val - h[shift]) > DEVIATION * POINT:
|
|
val = 0.0
|
|
else:
|
|
for back in range(1, BACKSTEP + 1):
|
|
res = hm[shift - back]
|
|
if res != 0.0 and res < val:
|
|
hm[shift - back] = 0.0
|
|
hm[shift] = val if h[shift] == val else 0.0
|
|
# --- final selection
|
|
if extreme_search == 0:
|
|
last_low = last_high = 0.0
|
|
else:
|
|
last_low, last_high = curlow, curhigh
|
|
last_low_pos = last_high_pos = 0
|
|
for shift in range(start, rates_total):
|
|
if extreme_search == 0:
|
|
if last_low == 0.0 and last_high == 0.0:
|
|
if hm[shift] != 0.0:
|
|
last_high = h[shift]
|
|
last_high_pos = shift
|
|
extreme_search = -1
|
|
zz[shift] = last_high
|
|
if lm[shift] != 0.0:
|
|
last_low = l[shift]
|
|
last_low_pos = shift
|
|
extreme_search = 1
|
|
zz[shift] = last_low
|
|
elif extreme_search == 1: # Peak: still updating the low, waiting for a high
|
|
if lm[shift] != 0.0 and lm[shift] < last_low and hm[shift] == 0.0:
|
|
zz[last_low_pos] = 0.0
|
|
last_low_pos = shift
|
|
last_low = lm[shift]
|
|
zz[shift] = last_low
|
|
if hm[shift] != 0.0 and lm[shift] == 0.0:
|
|
last_high = hm[shift]
|
|
last_high_pos = shift
|
|
zz[shift] = last_high
|
|
extreme_search = -1
|
|
else: # Bottom
|
|
if hm[shift] != 0.0 and hm[shift] > last_high and lm[shift] == 0.0:
|
|
zz[last_high_pos] = 0.0
|
|
last_high_pos = shift
|
|
last_high = hm[shift]
|
|
zz[shift] = last_high
|
|
if lm[shift] != 0.0 and hm[shift] == 0.0:
|
|
last_low = lm[shift]
|
|
last_low_pos = shift
|
|
zz[shift] = last_low
|
|
extreme_search = 1
|
|
self.prev = rates_total
|
|
return start
|
|
|
|
|
|
def main():
|
|
bk = fills.Book(SYM)
|
|
f = book.frame(SYM, "H1", bk)
|
|
a = f.atr(14)
|
|
sp = float(np.nanmean(f.spread))
|
|
n = f.n
|
|
print(f"{SYM} H1: {n} bars | stock ZigZag({DEPTH},{DEVIATION},{BACKSTEP}) | mean spread {sp:.2f} pts")
|
|
|
|
# ---------- 1. FINAL swings: one full pass over all history ----------
|
|
zfin = ZigZag(f.h, f.l, n)
|
|
zfin.calculate(n)
|
|
fp = np.flatnonzero(zfin.zz)
|
|
fv = zfin.zz[fp]
|
|
amp = np.abs(np.diff(fv))
|
|
dur = np.diff(fp)
|
|
amp_atr = amp / a[fp[1:]]
|
|
print(f"\nFINAL swings: {len(amp)} legs ({1000.0 * len(amp) / n:.1f}/1000 bars), "
|
|
f"median duration {np.median(dur):.0f} bars")
|
|
print(f" 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")
|
|
print(f" perfect foresight (untradeable): {amp.mean() - sp:+.1f} pts/leg net of spread")
|
|
|
|
# ---------- 2. LIVE replay: indicator state bar by bar, trade the drawn pivot ----------
|
|
z = ZigZag(f.h, f.l, n)
|
|
z.calculate(INIT_BARS)
|
|
pos_dir = np.zeros(n, np.int8) # +1 long, -1 short, decided at each bar close
|
|
newest_piv = np.full(n, -1, np.int64)
|
|
for i in range(INIT_BARS, n):
|
|
z.calculate(i + 1)
|
|
j = i
|
|
while j >= 0 and z.zz[j] == 0.0:
|
|
j -= 1
|
|
if j < 0:
|
|
continue
|
|
newest_piv[i] = j
|
|
pos_dir[i] = +1 if z.zz[j] == f.l[j] else -1 # newest LOW -> leg up -> long
|
|
# repaint audit: how many once-drawn newest pivots survive into the final line?
|
|
drawn = np.unique(newest_piv[newest_piv >= 0])
|
|
final_set = set(fp.tolist())
|
|
gone = np.array([p not in final_set for p in drawn])
|
|
print(f"\nLIVE replay: {len(drawn)} distinct newest-pivots were drawn; "
|
|
f"{gone.sum()} ({100.0 * gone.mean():.0f}%) later repainted away entirely")
|
|
|
|
# trades = position flips, filled at the next M1's real ask/bid
|
|
flips = np.flatnonzero((pos_dir[1:] != pos_dir[:-1]) & (pos_dir[1:] != 0) & (pos_dir[:-1] != 0)) + 1
|
|
if len(flips) < 10:
|
|
print(" too few flips")
|
|
return
|
|
ei = np.minimum(flips + 1, n - 1)
|
|
e = f.i0[ei] # entry minute of each new position
|
|
side = pos_dir[flips].astype(np.float64)
|
|
e_in, e_out = e[:-1], e[1:] # each position runs flip -> next flip
|
|
s = side[:-1]
|
|
entry_mid = 0.5 * (bk.ao[e_in] + bk.bo[e_in])
|
|
exit_mid = 0.5 * (bk.ao[e_out] + bk.bo[e_out])
|
|
gross = s * (exit_mid - entry_mid)
|
|
net = np.where(s > 0, bk.bo[e_out] - bk.ao[e_in], bk.bo[e_in] - bk.ao[e_out])
|
|
hold = (flips[1:] - flips[:-1])
|
|
print(f" policy: hold the newest drawn pivot's direction | {len(gross)} trades, "
|
|
f"median hold {np.median(hold):.0f} bars")
|
|
for name, m in (("ALL", np.ones(len(gross), bool)), ("LONG", s > 0), ("SHORT", s < 0)):
|
|
g, nn = gross[m], net[m]
|
|
se = g.std() / np.sqrt(len(g))
|
|
print(f" {name:<5} n={len(g):5d}: GROSS {g.mean():+7.2f} pts/trade (t={g.mean() / se:+.1f}) | "
|
|
f"NET {nn.mean():+7.2f} | win {100.0 * (nn > 0).mean():.0f}%")
|
|
print(f"\n (final-swing mean {amp.mean():.1f} pts vs what the live line hands over: see GROSS)")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|