divergence-signal-alerts/tests/reference.py

# SPDX-License-Identifier: LicenseRef-PolyForm-Noncommercial-1.0.0 (see LICENSE)
"""Python twin of mql5/Include/Divergence/Div.mqh: the same two oscillators (RSI, MACD), pivots and divergence rule, operation for
operation, so MT5's output can be compared bar by bar (tests/check_parity.py) and frequencies measured in Python.

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

import numpy as np

RSI, MACD = range(2)
NAMES = ["RSI", "MACD"]


@dataclass
class Params:
    use: tuple = (True, True)
    rsi_n: int = 14
    macd_fast: int = 12
    macd_slow: int = 26
    macd_sig: int = 9
    left: int = 5
    right: int = 3
    min_dist: int = 5
    max_dist: int = 60
    regular: bool = True
    hidden: bool = False
    clean: bool = True
    min_count: int = 1


class Ema:
    """SMA-seeded EMA (DVEmaStep)."""

    def __init__(self):
        self.cnt, self.sum, self.v = 0, 0.0, None

    def step(self, n, x):
        self.cnt += 1
        if self.cnt < n:
            self.sum += x
            self.v = None
        elif self.cnt == n:
            self.sum += x
            self.v = self.sum / n
        else:
            self.v = self.v + 2.0 / (n + 1) * (x - self.v)
        return self.v


def oscillators(c, p: Params):
    """Both oscillators on every bar (None until they have enough bars), as DVOscStep computes them."""
    n = len(c)
    out = [[None] * n for _ in range(2)]
    sg = sl = gain = loss = 0.0
    ef, es, esg = Ema(), Ema(), Ema()
    for k in range(n):
        # RSI (Wilder)
        if k >= 1:
            ch = c[k] - c[k - 1]
            g, ls = max(ch, 0.0), max(-ch, 0.0)
            m = p.rsi_n
            if k < m:
                sg += g
                sl += ls
            elif k == m:
                sg += g
                sl += ls
                gain, loss = sg / m, sl / m
            else:
                gain = (gain * (m - 1) + g) / m
                loss = (loss * (m - 1) + ls) / m
            if k >= m:
                out[RSI][k] = 100.0 if loss == 0.0 else (0.0 if gain == 0.0 else 100.0 - 100.0 / (1.0 + gain / loss))
        # MACD histogram
        a, b = ef.step(p.macd_fast, c[k]), es.step(p.macd_slow, c[k])
        if a is not None and b is not None:
            mline = a - b
            s = esg.step(p.macd_sig, mline)
            if s is not None:
                out[MACD][k] = mline - s
    return out


def pivot_low(l, p, pr):
    if p - pr.left < 0:
        return False
    v = l[p]
    return all(v < l[p - j] for j in range(1, pr.left + 1)) and all(not (v > l[p + j]) for j in range(1, pr.right + 1))


def pivot_high(h, p, pr):
    if p - pr.left < 0:
        return False
    v = h[p]
    return all(v > h[p - j] for j in range(1, pr.left + 1)) and all(not (v < h[p + j]) for j in range(1, pr.right + 1))


def clean(series, q, p, d):
    a, b = series[q], series[p]
    for j in range(q + 1, p):
        line = a + (b - a) * (j - q) / float(p - q)
        v = series[j]
        if v is None:
            return False
        if (v < line) if d > 0 else (v > line):
            return False
    return True


def check(h, l, osc, pr, p, d, prev):
    """DVCheck: the first previous pivot q with enough diverging oscillators; None if none."""
    price = l if d > 0 else h
    for q in prev:
        dist = p - q
        if dist < pr.min_dist:
            continue
        if dist > pr.max_dist:
            break
        pp, pq = price[p], price[q]
        if d > 0:
            typ = 1 if pp < pq else (2 if pp > pq else 0)
        else:
            typ = 1 if pp > pq else (2 if pp < pq else 0)
        if typ == 0 or (typ == 1 and not pr.regular) or (typ == 2 and not pr.hidden):
            continue
        if pr.clean and not clean(price, q, p, d):
            continue
        mask = cnt = 0
        for o in range(2):
            if not pr.use[o]:
                continue
            op, oq = osc[o][p], osc[o][q]
            if op is None or oq is None:
                continue
            div = d * (op - oq) > 0 if typ == 1 else d * (op - oq) < 0
            if not div:
                continue
            if pr.clean and not clean(osc[o], q, p, d):
                continue
            mask |= 1 << o
            cnt += 1
        if cnt >= max(1, pr.min_count):
            return dict(dir=d, type=typ, mask=mask, count=cnt, p=p, q=q, ppx=pp, qpx=pq)
    return None


def run(h, l, c, pr: Params, start: int = 0, stop: int | None = None):
    """Feed bars start..stop-1 (closed bars) as the indicator does; events keyed by the confirming bar index."""
    stop = len(c) if stop is None else stop
    h, l, c = (list(map(float, x[start:stop])) for x in (h, l, c))
    osc = oscillators(c, pr)
    pl, ph = [], []
    events = {}
    for k in range(len(c)):
        p = k - pr.right
        if p < 0:
            continue
        ev = None
        if pivot_low(l, p, pr):
            ev = check(h, l, osc, pr, p, 1, pl)
            pl.insert(0, p)
            del pl[4:]
        is_ph = pivot_high(h, p, pr)
        if ev is None and is_ph:
            ev = check(h, l, osc, pr, p, -1, ph)
        if is_ph:
            ph.insert(0, p)
            del ph[4:]
        if ev is not None:
            ev = {**ev, "p": ev["p"] + start, "q": ev["q"] + start}
            events[k + start] = ev
    full = [[None] * start + o for o in osc]
    return events, full