harmonic-signal-alerts/tests/reference.py

# SPDX-License-Identifier: LicenseRef-PolyForm-Noncommercial-1.0.0 (see LICENSE)
"""Python twin of Harm.mqh: the same swings, pattern ratios and PRZ completion rule, step for step, so the indicators'
output can be compared bar by bar (tests/check_parity.py). The rules are described in the README.

    from reference import Params, run
    events = run(high, low, close, Params())
"""
from dataclasses import dataclass

ABCD, GARTLEY, BAT = range(3)
NAMES = ["ABCD", "Gartley", "Bat"]
# b_lo, b_hi (AB/XA), c_lo, c_hi (BC/AB), d (key ratio), p_lo, p_hi (CD/BC at the key level, 0 = no check)
RATIOS = [
    (0, 0, 0.382, 0.886, 1.0, 0, 0),
    (0.618, 0.618, 0.382, 0.886, 0.786, 1.13, 1.618),
    (0.382, 0.50, 0.382, 0.886, 0.886, 1.618, 2.618),
]


@dataclass
class Params:
    left: int = 10
    right: int = 5
    tol: float = 0.05
    prz: float = 0.03
    use: tuple = (True, True, True)


def inside(v, lo, hi, tol):
    return lo * (1 - tol) <= v <= hi * (1 + tol)


def match(sw, pr):
    """Patterns completed by the last swing (C) of `sw` [(type, price, bar)]: list of candidate dicts."""
    out = []
    if len(sw) < 3:
        return out
    ct, cp, cb = sw[-1]
    bt, bp, bb = sw[-2]
    at, ap, ab = sw[-3]
    bull = ct == 1                       # C is a high: D below it
    s = 1.0 if bull else -1.0
    AB = abs(ap - bp)
    BC = abs(cp - bp)
    if AB <= 0 or BC <= 0:
        return out
    for pat in range(len(NAMES)):
        if not pr.use[pat]:
            continue
        b_lo, b_hi, c_lo, c_hi, d, p_lo, p_hi = RATIOS[pat]
        if pat == ABCD:
            if not inside(BC / AB, c_lo, c_hi, pr.tol):
                continue
            key = cp - s * AB
            scale = AB
            xp, xb = None, None
        else:
            if len(sw) < 4:
                continue
            xt, xp, xb = sw[-4]
            XA = abs(ap - xp)
            if XA <= 0 or not inside(AB / XA, b_lo, b_hi, pr.tol):
                continue
            if not inside(BC / AB, c_lo, c_hi, pr.tol):
                continue
            key = ap - s * d * XA
            if p_lo > 0 and not inside(abs(cp - key) / BC, p_lo, p_hi, pr.tol):
                continue
            scale = XA
        if s * (cp - key) <= 0:
            continue
        out.append(dict(pat=pat, dir=1 if bull else -1,
                        near=key + s * pr.prz * scale, far=key - s * pr.prz * scale,
                        x=(xb, xp), a=(ab, ap), b=(bb, bp), c=(cb, cp), ext=None, ext_bar=-1, worst=None))
    return out


def run(high, low, close, pr: Params = Params()):
    """events: list of dicts in firing order; each has the bar it fired on, the points and the PRZ."""
    n = len(high)
    sw = []                               # (type 1 high / -1 low, price, bar)
    cands = []
    events = []
    for t in range(n):
        h, l, c = high[t], low[t], close[t]
        # 1) live candidates take this bar
        alive = []
        for cd in cands:
            if step(cd, t, h, l, c, events):
                alive.append(cd)
        cands = alive
        # 2) pivot confirmed at p = t - right
        p = t - pr.right
        if p - pr.left < 0:
            continue
        is_h = all(high[p] > high[p - j] for j in range(1, pr.left + 1)) and \
            all(high[p] >= high[p + j] for j in range(1, pr.right + 1))
        is_l = all(low[p] < low[p - j] for j in range(1, pr.left + 1)) and \
            all(low[p] <= low[p + j] for j in range(1, pr.right + 1))
        if is_h and is_l:
            if not sw:
                continue
            is_h, is_l = sw[-1][0] == -1, sw[-1][0] == 1
        changed = False
        for typ, ok, px in ((1, is_h, high[p]), (-1, is_l, low[p])):
            if not ok:
                continue
            if sw and sw[-1][0] == typ:
                if (px > sw[-1][1]) if typ == 1 else (px < sw[-1][1]):
                    sw[-1] = (typ, px, p)
                    changed = True
            elif sw and ((typ == -1 and px >= sw[-1][1]) or (typ == 1 and px <= sw[-1][1])):
                pass                                     # on the wrong side of the last swing
            else:
                sw.append((typ, px, p))
                if len(sw) > 8:
                    sw.pop(0)
                changed = True
        if not changed:
            continue
        cands = match(sw, pr)
        # 3) new candidates catch up on the bars after C (C's pivot guarantees none of them is beyond C)
        alive = []
        for cd in cands:
            ok = True
            for j in range(cd["c"][0] + 1, t + 1):
                if not step(cd, j, high[j], low[j], close[j], events, fire_at=t):
                    ok = False
                    break
            if ok:
                alive.append(cd)
        cands = alive
    return events


def step(cd, j, h, l, c, events, fire_at=None):
    """Feed bar j to a candidate. False when it is finished (dead, void or fired). `fire_at` = the bar an event is
    dated on during the catch-up (the confirmation bar, never earlier)."""
    s = cd["dir"]
    if (h > cd["c"][1]) if s > 0 else (l < cd["c"][1]):
        return False                                     # beyond C: C moves, the candidate dies
    ext = l if s > 0 else h
    if cd["ext"] is None or ((ext < cd["ext"]) if s > 0 else (ext > cd["ext"])):
        cd["ext"], cd["ext_bar"] = ext, j
    if cd["worst"] is None or ((c < cd["worst"]) if s > 0 else (c > cd["worst"])):
        cd["worst"] = c
    if (cd["ext"] <= cd["near"]) if s > 0 else (cd["ext"] >= cd["near"]):
        if (cd["worst"] < cd["far"]) if s > 0 else (cd["worst"] > cd["far"]):
            return False                                 # closed beyond the zone: void
        bar = j if fire_at is None else fire_at
        events.append(dict(bar=bar, dir=s, pat=cd["pat"], name=NAMES[cd["pat"]], x=cd["x"], a=cd["a"], b=cd["b"],
                           c=cd["c"], d=(cd["ext_bar"], cd["ext"]), near=cd["near"], far=cd["far"]))
        return False
    return True