Warrior_EA/research/classic.py

"""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