candlestick-signal-alerts/tests/reference.py

# SPDX-License-Identifier: LicenseRef-PolyForm-Noncommercial-1.0.0 (see LICENSE)
"""Python twin of Cdl.mqh: 13 candlestick patterns on closed bars (the rules are in the README). This file is the exact
rule book; the MQL5 and MQL4 engines are checked against it bar by bar (tests/check_parity.py).

    from reference import run, NAMES
    rows = run(o, h, l, c)         # rows[k] = dict(ev=[(pattern, dir), ...], mask=bit mask, ema=EMA(20) or None)
"""
NAMES = ["Bullish Engulfing", "Bearish Engulfing", "Hammer", "Hanging Man", "Inverted Hammer", "Shooting Star",
         "Doji", "Morning Star", "Evening Star", "Bullish Harami", "Bearish Harami", "Piercing Line",
         "Dark Cloud Cover"]
NP = len(NAMES)
ALL = (True,) * NP
# fixed direction per pattern (Doji: by the trend before it, 0 here)
DIR = [1, -1, 1, -1, 1, -1, 0, 1, -1, 1, -1, 1, -1]
EMA_N, AVG_N = 20, 10


def run(o, h, l, c, use=ALL):
    n = len(c)
    ema = [None] * n
    s, cnt = 0.0, 0
    for k in range(n):                         # EMA(20) of closes, SMA-seeded
        cnt += 1
        if cnt < EMA_N:
            s += c[k]
        elif cnt == EMA_N:
            s += c[k]
            ema[k] = s / EMA_N
        else:
            ema[k] = ema[k - 1] + 2.0 / (EMA_N + 1) * (c[k] - ema[k - 1])
    body = [abs(c[k] - o[k]) for k in range(n)]
    rng = [h[k] - l[k] for k in range(n)]
    up = [h[k] - max(o[k], c[k]) for k in range(n)]
    lo = [min(o[k], c[k]) - l[k] for k in range(n)]
    bull = [c[k] > o[k] for k in range(n)]
    bear = [c[k] < o[k] for k in range(n)]

    def avg(arr, k):                           # SMA of the AVG_N bars before k
        return sum(arr[k - AVG_N:k]) / AVG_N if k >= AVG_N else None

    rows = []
    for k in range(n):
        ev = []                                 # (pattern, dir)
        ab, ar = avg(body, k), avg(rng, k)
        if ab is not None and k >= 3:
            b, r, u, w = body[k], rng[k], up[k], lo[k]
            tol = 1e-9 * abs(c[k])                # ties within 1e-9 x price count as equal (rounding noise decides nothing)

            def GE(x, y):
                return x >= y - tol

            def LE(x, y):
                return x <= y + tol

            def GT(x, y):
                return x > y + tol

            def LT(x, y):
                return x < y - tol

            def tr(j):                          # +1 up / -1 down / 0 unknown, at bar j
                if j < 0 or ema[j] is None:
                    return 0
                return 1 if GT(c[j], ema[j]) else (-1 if LT(c[j], ema[j]) else 0)

            t1, t2, t3 = tr(k - 1), tr(k - 2), tr(k - 3)
            p = k - 1
            # 0/1 engulfing
            if bear[p] and bull[k] and LE(o[k], c[p]) and GE(c[k], o[p]) and GT(b, body[p]) and t2 < 0:
                ev.append((0, 1))
            if bull[p] and bear[k] and GE(o[k], c[p]) and LE(c[k], o[p]) and GT(b, body[p]) and t2 > 0:
                ev.append((1, -1))
            if GT(r, 0):
                hammer = GE(w, 2 * b) and LE(u, 0.1 * r) and GE(w, 0.6 * r)
                inv = GE(u, 2 * b) and LE(w, 0.1 * r) and GE(u, 0.6 * r)
                if hammer and t1 < 0:
                    ev.append((2, 1))
                if hammer and t1 > 0:
                    ev.append((3, -1))
                if inv and t1 < 0:
                    ev.append((4, 1))
                if inv and t1 > 0:
                    ev.append((5, -1))
                if LE(b, 0.05 * r) and GE(r, 0.5 * ar) and t1 != 0:
                    ev.append((6, -t1))
            # 7/8 stars
            q = k - 2
            if bear[q] and GE(body[q], ab) and LE(body[p], 0.3 * body[q]) and LE(min(o[p], c[p]), c[q]) and bull[k] \
                    and GT(c[k], (o[q] + c[q]) / 2) and t3 < 0:
                ev.append((7, 1))
            if bull[q] and GE(body[q], ab) and LE(body[p], 0.3 * body[q]) and GE(max(o[p], c[p]), c[q]) and bear[k] \
                    and LT(c[k], (o[q] + c[q]) / 2) and t3 > 0:
                ev.append((8, -1))
            # 9/10 harami
            if bear[p] and GE(body[p], ab) and bull[k] and GE(o[k], c[p]) and LE(c[k], o[p]) and LT(b, body[p]) and t2 < 0:
                ev.append((9, 1))
            if bull[p] and GE(body[p], ab) and bear[k] and LE(o[k], c[p]) and GE(c[k], o[p]) and LT(b, body[p]) and t2 > 0:
                ev.append((10, -1))
            # 11/12 piercing, dark cloud
            if bear[p] and GE(body[p], ab) and bull[k] and LE(o[k], c[p]) and GT(c[k], (o[p] + c[p]) / 2) and LT(c[k], o[p]) and t2 < 0:
                ev.append((11, 1))
            if bull[p] and GE(body[p], ab) and bear[k] and GE(o[k], c[p]) and LT(c[k], (o[p] + c[p]) / 2) and GT(c[k], o[p]) and t2 > 0:
                ev.append((12, -1))
        ev = [e for e in ev if use[e[0]]]
        ev.sort()
        rows.append(dict(ev=ev, mask=sum(1 << pt for pt, _ in ev), ema=ema[k]))
    return rows