animatedread/Warrior_EA
Watch
4
6
Fork
You've already forked Warrior_EA
 7
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
Warrior_EA/research/classic.py
AnimateDread 8710240cd5 fix(signals): revive a dead MA model, and demote Sanyaku from state to event 
Two defects surfaced by research/test_classic.py, both verified fixed by re-running the
transcription against 178k bars of EURUSD H1.

CSignalMA model 1 could never fire. For any recursive average - and MA_TYPE_EMA is the
shipped default - MA(i) = a*Close(i) + (1-a)*MA(i+1), so

    DiffMA(i)      = a     * (Close(i) - MA(i+1))
    DiffCloseMA(i) = (1-a) * (Close(i) - MA(i+1))

are positive multiples of one quantity and always share a sign. Model 1 asks for a close
BELOW a RISING average, which is precisely the combination that identity forbids: 0.000%
of bars, either direction, any symbol. The MQL5 standard library this was ported from
defaults to MODE_SMA, where the two are merely correlated - the bug arrived with the EMA
default, not with the port. Reading the slope one bar back (DiffMAPrev) breaks the tie for
every MA type while keeping the model's stated meaning. Now fires on 7.92% of bars.

CSignalIchimoku model 11 fired on 27% of bars at weight 100. Sanyaku is three standing
STATES conjoined with no transition term, so it held across long stretches - and being
last in the if-chain at the top weight, the module's highest-conviction reading was also
its most common one, overwriting all eight event models below it on a quarter of all bars.
The old comment rejected an event form because "demanding all three flip on the same bar
would fire almost never" - true, but that is not the alternative. Kouten is the TURN: the
ALIGNMENT transitions, and only one role need change for it to. Testing !Sanyaku(idx+1)
fires once per aligned stretch. Now 2.17%, in line with Kumo breakout (2.4%) and the
strong TK cross (1.1%). DataReady() extended one bar deeper to cover the lookback.

Neither pattern showed edge before or after; this is about the models meaning what they
say and the vote not being dominated by a constant.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-01 17:14:34 -04:00

376 lines
15 KiB
Python
Raw Permalink Blame History

"""Faithful Python transcription of the four classic vote modules in Signals/.
The point of this file is fidelity, NOT elegance. Every condition below is transcribed
literally from the .mqh it names, including quirks that look like bugs in the MQL5
standard library (see CompareMaps). We are testing what the EA actually does, so a
"corrected" version here would be testing something else.
Index convention: MQL5 series are newest-first (index 0 = current bar, +1 = older).
Numpy arrays here are oldest-first. So MQL5 `ind + k` == python `i - k`.
Decision timing: Warrior runs with Expert_EveryTick=false, so StartIndex() == 1 - the
last CLOSED bar. Every pattern below is therefore evaluated on bar i, and the resulting
trade is entered at the OPEN of bar i+1. Nothing reads beyond bar i.
Weights are the shipped constructor defaults. Patterns at weight 10 are the
"confirming" models: standing states rather than events, never to be acted on alone.
"""
import numpy as np
# --------------------------------------------------------------------------- indicators
def ema(x, n):
a = 2.0 / (n + 1.0)
out = np.empty_like(x, dtype=float)
out[0] = x[0]
for i in range(1, len(x)):
out[i] = a * x[i] + (1 - a) * out[i - 1]
return out
def rsi_wilder(c, n=14):
"""MT5's iRSI: Wilder smoothing, seeded with a simple average of the first n deltas."""
d = np.diff(c, prepend=c[0])
up = np.where(d > 0, d, 0.0)
dn = np.where(d < 0, -d, 0.0)
au = np.empty_like(c, dtype=float)
ad = np.empty_like(c, dtype=float)
au[:n] = up[1:n + 1].mean() if len(up) > n else up.mean()
ad[:n] = dn[1:n + 1].mean() if len(dn) > n else dn.mean()
for i in range(n, len(c)):
au[i] = (au[i - 1] * (n - 1) + up[i]) / n
ad[i] = (ad[i - 1] * (n - 1) + dn[i]) / n
return np.where(ad > 0, 100.0 - 100.0 / (1.0 + au / np.where(ad > 0, ad, 1e-12)), 100.0)
def macd(c, fast=12, slow=26, sig=9):
main = ema(c, fast) - ema(c, slow)
signal = ema(main, sig)
return main, signal
def rolling_max(x, n):
out = np.full(len(x), np.nan)
for i in range(n - 1, len(x)):
out[i] = x[i - n + 1:i + 1].max()
return out
def rolling_min(x, n):
out = np.full(len(x), np.nan)
for i in range(n - 1, len(x)):
out[i] = x[i - n + 1:i + 1].min()
return out
def ichimoku(h, l, c, pt=9, pk=26, ps=52):
"""Returns the RAW (unshifted) buffers, exactly as CiIchimoku exposes them.
SenkouSpan A/B at index i are computed FROM bar i and DRAWN pk bars ahead. The EA's
SpanAAt(ind) = SenkouSpanA(ind + pk) is therefore 'the cloud plotted at bar ind',
which in oldest-first indexing is raw[i - pk]. FutureSpan*(ind) is raw[i] - the
projection, which a live bar genuinely has. Keeping the raw buffers here lets the
pattern code below make that distinction explicitly instead of hiding it.
"""
tenkan = (rolling_max(h, pt) + rolling_min(l, pt)) / 2.0
kijun = (rolling_max(h, pk) + rolling_min(l, pk)) / 2.0
span_a_raw = (tenkan + kijun) / 2.0
span_b_raw = (rolling_max(h, ps) + rolling_min(l, ps)) / 2.0
return tenkan, kijun, span_a_raw, span_b_raw
def shift_older(x, k):
"""Value of x as it stood k bars ago; NaN where that predates the array."""
out = np.full(len(x), np.nan)
if k < len(x):
out[k:] = x[:len(x) - k] if k > 0 else x
return out if k > 0 else x.astype(float).copy()
# ------------------------------------------------------- oscillator extremum bit-map
# Transcribed from CSignalRSI::StateRSI / ExtStateRSI / CompareMaps (SignalRSI.mqh:155-322)
# and the identical machinery in CSignalMACD. NOTE the bit semantics: for BOTH minima
# and maxima the code sets bit 0 when the PREVIOUS extremum is further from the current
# one in the "less extreme" direction. The standard library's own header comment claims
# the opposite polarity. We transcribe the CODE, not the comment.
def turning_points(osc):
"""Indices where the oscillator changes direction, with type (+1 = minimum, -1 = maximum).
StateRSI walks back from `ind` while diff keeps one sign and stops at the reversal;
precomputing every turning point once and bisecting is the same answer, ~1000x faster.
"""
d = np.diff(osc)
s = np.sign(d)
s[s == 0] = 0
idx, typ = [], []
for i in range(1, len(s)):
if s[i] != 0 and s[i - 1] != 0 and s[i] != s[i - 1]:
# direction changed AT bar i: rising->falling = maximum, falling->rising = minimum
idx.append(i)
typ.append(-1 if s[i] < 0 else +1)
return np.array(idx, dtype=int), np.array(typ, dtype=int)
def _price_extremum(h, l, pos, is_min, first_two, cap=None):
"""m_extr_pr[i]: MinValue(pos-2,5) / MaxValue(pos-2,5), or a 4-bar window for i<=1.
MQL5 MinValue(start,count) scans series indices start..start+count-1, i.e. from
NEWER to OLDER. start=pos-2 with count=5 is therefore the 5 bars CENTRED on pos -
it reads up to 2 bars NEWER than the extremum it is describing.
That is a lookahead in the standard library itself: the most recent extremum can sit
within 2 bars of the decision bar, so the window reaches past it into bars that have
not closed. `cap` clamps the window to the decision bar, which is what a live EA can
actually see. Pass cap=None to reproduce the leaky original.
"""
lo = max(pos - 2, 0)
hi = pos + 2 if not first_two else pos + 1
if cap is not None:
hi = min(hi, cap)
hi = min(hi, len(h) - 1)
if lo > hi:
return np.nan
return l[lo:hi + 1].min() if is_min else h[lo:hi + 1].max()
def ext_state_map(osc, h, l, tp_idx, tp_typ, i, causal=True):
"""Build m_extr_map for bar i. Returns None if fewer than 3 extremums are reachable."""
# A turn AT bar p is only knowable once bar p+1 has closed, so at decision bar i the
# newest usable extremum is at p <= i-1. 'left' gives the last tp strictly below i.
# (MQL5's StateRSI has this right for free: it only ever walks backward from ind, so
# the extremum it stops on is always confirmed by a bar it has already seen.)
k = np.searchsorted(tp_idx, i, side='left') - 1
if k < 2:
return None
n_take = min(10, k + 1)
extr_map = 0
pr, oc = [], []
cap = i if causal else None
for j in range(n_take):
pos = tp_idx[k - j]
is_min = tp_typ[k - j] > 0
oc.append(osc[pos])
pr.append(_price_extremum(h, l, pos, is_min, j <= 1, cap))
if j > 1:
m = 0
if is_min:
if pr[j - 2] < pr[j]:
m += 1
if oc[j - 2] < oc[j]:
m += 4
else:
if pr[j - 2] > pr[j]:
m += 1
if oc[j - 2] > oc[j]:
m += 4
extr_map += m << (4 * (j - 2))
return extr_map
def compare_maps(extr_map, pattern, count, start=0):
"""CSignalRSI::CompareMaps - bit-for-bit."""
step = 8
total = step * (start + count)
if total > 32:
return False
i = step * start
j = 0
while i < total:
inp = (pattern >> j) & 3
if inp < 2:
if inp != ((extr_map >> i) & 3):
return False
inp = (pattern >> (j + 2)) & 3
if inp < 2:
if inp != ((extr_map >> (i + 2)) & 3):
return False
i += step
j += 4
return True
def divergence_flags(osc, h, l, gate_long, gate_short, causal=True):
"""Patterns 'divergence' (CompareMaps(1,1)) and 'double divergence' (CompareMaps(0x11,2)).
gate_long/gate_short are the caller's precondition masks (e.g. MACD requires
Main < 0 for longs), so the expensive map build only runs where a pattern could fire.
"""
n = len(osc)
tp_idx, tp_typ = turning_points(osc)
d1L = np.zeros(n, bool); d2L = np.zeros(n, bool)
d1S = np.zeros(n, bool); d2S = np.zeros(n, bool)
need = gate_long | gate_short
for i in np.nonzero(need)[0]:
m = ext_state_map(osc, h, l, tp_idx, tp_typ, i, causal)
if m is None:
continue
a = compare_maps(m, 1, 1)
b = compare_maps(m, 0x11, 2)
if gate_long[i]:
d1L[i] = a; d2L[i] = b
if gate_short[i]:
d1S[i] = a; d2S[i] = b
return d1L, d2L, d1S, d2S
# --------------------------------------------------------------------------- patterns
MODULES = {
'MA': [10, 10, 60, 60],
'RSI': [10, 60, 80, 100],
'MACD': [10, 30, 80, 50, 60, 100],
'Ichimoku': [10, 10, 10, 10, 30, 40, 60, 70, 80, 90, 90, 100],
}
def build_patterns(o, h, l, c, causal=True):
"""Returns (fireL, fireS, names, weights) - boolean [n, 26] matrices.
causal=True clamps the divergence price-extremum window to the decision bar. See
_price_extremum: the shipped standard-library version reads up to 2 bars past it.
"""
n = len(c)
L, S, names, weights = [], [], [], []
def add(mod, num, longmask, shortmask):
L.append(longmask); S.append(shortmask)
names.append(f"{mod}_p{num}"); weights.append(MODULES[mod][num])
prev = lambda x: shift_older(x, 1)
# ---------------- CSignalMA (SignalMA.mqh:170-300), EMA(12) on close
ma = ema(c, 12)
dma = ma - prev(ma)
dopen, dhigh, dlow, dclose = o - ma, h - ma, l - ma, c - ma
# LongCondition (SignalMA.mqh:179): `if(DiffCloseMA < 0)` -> p1 only; ELSE -> p0, then
# `if(DiffMA > 0)` splits on DiffOpenMA: < 0 gives p2 (roll-back cross), >= 0 gives p3
# (formed piercing, needs the low through the MA). ShortCondition (line 243) mirrors it.
# p1 uses the slope AS OF THE PREVIOUS BAR (DiffMAPrev). With a recursive average, DiffMA and
# DiffCloseMA are positive multiples of the same quantity and can never disagree in sign, which
# made "close below a rising MA" unsatisfiable - the model never fired at the shipped EMA default.
dma_prev = prev(ma) - shift_older(ma, 2)
add('MA', 0, (dclose >= 0), (dclose <= 0))
add('MA', 1, (dclose < 0) & (dopen > 0) & (dma_prev > 0),
(dclose > 0) & (dopen < 0) & (dma_prev < 0))
add('MA', 2, (dclose >= 0) & (dma > 0) & (dopen < 0),
(dclose <= 0) & (dma < 0) & (dopen > 0))
add('MA', 3, (dclose >= 0) & (dma > 0) & (dopen >= 0) & (dlow < 0),
(dclose <= 0) & (dma < 0) & (dopen <= 0) & (dhigh > 0))
# ---------------- CSignalRSI (SignalRSI.mqh:326-422), RSI(14) on close
r = rsi_wilder(c, 14)
dr = r - prev(r)
drp = prev(dr)
rp = prev(r)
upL, upS = dr > 0, dr < 0
d1L, d2L, d1S, d2S = divergence_flags(r, h, l, upL, upS, causal)
add('RSI', 0, upL, upS)
add('RSI', 1, upL & (drp < 0) & (rp < 30.0), upS & (drp > 0) & (rp > 70.0))
add('RSI', 2, d1L, d1S)
add('RSI', 3, d2L, d2S)
# ---------------- CSignalMACD (SignalMACD.mqh:352-465), 12/26/9 on close
main, sig = macd(c, 12, 26, 9)
dmain = main - prev(main)
dmainp = prev(dmain)
state = main - sig
statep = prev(state)
mainp = prev(main)
gL, gS = dmain > 0, dmain < 0
m1L, m2L, m1S, m2S = divergence_flags(main, h, l, gL & (main < 0), gS & (main > 0), causal)
add('MACD', 0, gL, gS)
add('MACD', 1, gL & (dmainp < 0), gS & (dmainp > 0))
add('MACD', 2, gL & (state > 0) & (statep < 0), gS & (state < 0) & (statep > 0))
add('MACD', 3, gL & (main > 0) & (mainp < 0), gS & (main < 0) & (mainp > 0))
add('MACD', 4, m1L, m1S)
add('MACD', 5, m2L, m2S)
# ---------------- CSignalIchimoku (SignalIchimoku.mqh:356-590), 9/26/52
PK = 26
tk, kj, sa_raw, sb_raw = ichimoku(h, l, c, 9, PK, 52)
sa = shift_older(sa_raw, PK) # cloud AS PLOTTED AT this bar
sb = shift_older(sb_raw, PK)
ctop, cbot = np.fmax(sa, sb), np.fmin(sa, sb)
ctop_p, cbot_p = prev(ctop), prev(cbot)
cp = prev(c)
kjp = prev(kj)
above, below = c > ctop, c < cbot
inside = (c >= cbot) & (c <= ctop)
xup = (tk > kj) & (prev(tk) <= prev(kj))
xdn = (tk < kj) & (prev(tk) >= prev(kj))
# Chikou: this close vs the close AND the cloud-top PK bars back - both strictly past.
c_pk = shift_older(c, PK)
ctop_pk, cbot_pk = shift_older(ctop, PK), shift_older(cbot, PK)
chiL = (c > c_pk) & (c > ctop_pk)
chiS = (c < c_pk) & (c < cbot_pk)
# thin cloud: this bar's thickness < half the mean thickness of the last PK bars
thick = np.abs(sa - sb)
mean_thick = np.convolve(np.nan_to_num(thick), np.ones(PK) / PK, mode='full')[:len(thick)]
thin = (mean_thick > 0) & (thick < 0.5 * mean_thick)
fa, fb = sa_raw, sb_raw # PROJECTED cloud (computed from this bar)
fap, fbp = prev(fa), prev(fb)
add('Ichimoku', 0, above, below)
add('Ichimoku', 1, fa > fb, fa < fb)
add('Ichimoku', 2, chiL & above, chiS & below)
add('Ichimoku', 3, xup & (c < cbot), xdn & (c > ctop))
add('Ichimoku', 4, (fa > fb) & (fap <= fbp), (fa < fb) & (fap >= fbp))
add('Ichimoku', 5, xup & inside, xdn & inside)
add('Ichimoku', 6, (c > kj) & (cp <= kjp), (c < kj) & (cp >= kjp))
add('Ichimoku', 7, above & (cp > kjp) & (l <= kj) & (c > kj),
below & (cp < kjp) & (h >= kj) & (c < kj))
add('Ichimoku', 8, above & (cp <= ctop_p), below & (cp >= cbot_p))
add('Ichimoku', 9, above & (cp <= ctop_p) & thin, below & (cp >= cbot_p) & thin)
add('Ichimoku', 10, xup & above, xdn & below)
# p11 Sanyaku is an EVENT: the bar the three-role alignment BECOMES true. As a standing
# conjunction it held on 27% of bars, so the module's weight-100 reading was also its most
# common one and it overwrote all eight event models below it in the if-chain.
sanL = (tk > kj) & chiL & above
sanS = (tk < kj) & chiS & below
p_sanL = np.nan_to_num(shift_older(sanL.astype(float), 1)).astype(bool)
p_sanS = np.nan_to_num(shift_older(sanS.astype(float), 1)).astype(bool)
add('Ichimoku', 11, sanL & ~p_sanL, sanS & ~p_sanS)
fireL = np.column_stack(L)
fireS = np.column_stack(S)
fireL = np.nan_to_num(fireL).astype(bool)
fireS = np.nan_to_num(fireS).astype(bool)
# warm-up: nothing is valid until every indicator has its full lookback
warm = 2 * PK + 52 + 5
fireL[:warm] = False
fireS[:warm] = False
return fireL, fireS, names, np.array(weights, dtype=float)
# --------------------------------------------------------------------------- voting
MODULE_SLICES = {'MA': (0, 4), 'RSI': (4, 8), 'MACD': (8, 14), 'Ichimoku': (14, 26)}
def module_votes(fireL, fireS, weights):
"""Per-module signed vote. Within a module the LAST matching pattern wins - the .mqh
if-chains assign `result = m_pattern_N` in ascending N, so the highest-numbered match
is what the module casts. Not a sum."""
n = fireL.shape[0]
votes = {}
for mod, (a, b) in MODULE_SLICES.items():
vl = np.zeros(n); vs = np.zeros(n)
for k in range(a, b):
vl = np.where(fireL[:, k], weights[k], vl)
vs = np.where(fireS[:, k], weights[k], vs)
votes[mod] = vl - vs
return votes
def combined_direction(votes):
"""CExpertSignalCustom::Direction() - average of the non-zero module votes."""
mods = list(votes.values())
tot = np.zeros(len(mods[0]))
cnt = np.zeros(len(mods[0]))
for v in mods:
tot += v
cnt += (v != 0)
out = np.where(cnt > 0, tot / np.where(cnt > 0, cnt, 1), 0.0)
out[np.abs(out) > 100] = 0.0 # the EA's own range check
return out
Reference in a new issue View git blame Copy permalink