# SPDX-License-Identifier: LicenseRef-PolyForm-Noncommercial-1.0.0 (see LICENSE) """Independent Python reference of the Displacement Order Blocks engine (mql5/.../DzEngine.mqh), bar by bar. Everything at bar i is decided from bars <= i; the last bar of the arrays is the forming bar and is never used. The rules are the ones in the DzEngine.mqh header (d = +1 demand, -1 supply): ATR Wilder (RMA) of the true range, seeded with the SMA of the first `atr` values (summed in order, as MQL5); the impulse test of bar i uses the ATR after bar i-1. swing left `swing` bars strictly lower highs (higher lows), right `swing` bars not above (not below); known at p + swing; a new swing replaces the previous one of its side. break the first close beyond the newest swing; it uses the swing up whether or not a zone is made. impulse range >= disp_mult x ATR, body >= body_pct % of the range (within 1e-9 x range, for exact decimal ties), closing in the break's direction. base base_mode 0: nearest candle of the other colour among the `search` bars before the break bar; base_mode 1: from the lowest low (demand; supply: highest high, latest bar on a tie) of those bars back, the nearest candle of the other colour. zone mode 0 full candle, 1 body, 2 mixed. states per bar for every live zone made before it: broken (checked first), then tested / retested. cap max_live live zones per side (0 = off): the oldest live one of that side is dropped (state 3). """ import numpy as np FRESH, TESTED, BROKEN, DROPPED = 0, 1, 2, 3 KEYS = ["dem_top", "dem_bot", "sup_top", "sup_bot", "break_up", "break_dn", "tested", "retests", "broken", "live_dem", "live_sup", "dem_base", "sup_base"] DEFAULTS = dict(swing=3, atr=14, search=12, zone_mode=0, test_mode=0, break_mode=0, max_live=12, base_mode=0, disp_mult=1.0, body_pct=55.0) def dz_zones(t, o, h, l, c, on_bar=None, **kw): """-> (buffers, zones, breaks, live zones, ATR after the last closed bar). on_bar(i, zones) runs after each bar.""" p = {**DEFAULTS, **kw} n = len(c) out = {k: np.full(n, np.nan) for k in KEYS} zones, breaks, live = [], [], [] sw = [dict(i=-1, p=0.0, broken=True), dict(i=-1, p=0.0, broken=True)] active = [0, 0] atr, atr_sum = None, 0.0 A, S = p["atr"], p["swing"] def reaches(z, hi, lo): tol = abs(z["mid"]) * 1e-10 m = p["test_mode"] if z["dir"] > 0: return lo <= z["top"] if m == 0 else (lo <= z["mid"] + tol if m == 1 else lo <= z["bot"]) return hi >= z["bot"] if m == 0 else (hi >= z["mid"] - tol if m == 1 else hi >= z["top"]) def breaks_zone(z, hi, lo, cl): if z["dir"] > 0: return cl < z["bot"] if p["break_mode"] == 0 else lo < z["bot"] return cl > z["top"] if p["break_mode"] == 0 else hi > z["top"] def is_swing(q, side): for k in range(1, S + 1): if side > 0 and (h[q - k] >= h[q] or h[q + k] > h[q]): return False if side < 0 and (l[q - k] <= l[q] or l[q + k] < l[q]): return False return True for i in range(n - 1): prev = atr tr = h[i] - l[i] if i == 0 else max(h[i] - l[i], max(abs(h[i] - c[i - 1]), abs(l[i] - c[i - 1]))) if i < A - 1: atr_sum += tr elif i == A - 1: atr_sum += tr atr = atr_sum / A else: atr = (atr * (A - 1) + tr) / A if prev is not None and prev > 0: n_test = n_re = n_brk = 0 keep = [] for z in live: reach = reaches(z, h[i], l[i]) if breaks_zone(z, h[i], l[i], c[i]): z["state"], z["end"] = BROKEN, i n_brk += 1 active[0 if z["dir"] > 0 else 1] -= 1 continue if reach: if z["state"] == FRESH: z["state"], z["retests"] = TESTED, 1 n_test += 1 elif not z["inside"]: z["retests"] += 1 n_re += 1 z["inside"] = reach keep.append(z) live = keep out["tested"][i], out["retests"][i], out["broken"][i] = n_test, n_re, n_brk for d in (1, -1): s = 0 if d > 0 else 1 if sw[s]["broken"] or c[i] * d <= sw[s]["p"] * d: continue sw[s]["broken"] = True out["break_up" if d > 0 else "break_dn"][i] = sw[s]["p"] rng, body = h[i] - l[i], abs(c[i] - o[i]) impulse = rng > 0 and rng >= p["disp_mult"] * prev and body >= p["body_pct"] / 100.0 * rng - rng * 1e-9 \ and (c[i] - o[i]) * d > 0 base, lo_j, from_j = -1, max(0, i - p["search"]), i - 1 if impulse and p["base_mode"] == 1: for j in range(i - 2, lo_j - 1, -1): # the move's start: the lowest low (highest high), latest on a tie if (l[j] < l[from_j]) if d > 0 else (h[j] > h[from_j]): from_j = j if impulse: for j in range(from_j, lo_j - 1, -1): if (c[j] - o[j]) * d < 0: base = j break zi = -1 if base >= 0: bt, bb = max(o[base], c[base]), min(o[base], c[base]) zm = p["zone_mode"] top = bt if zm == 1 else (bt if zm == 2 and d > 0 else h[base]) bot = bb if zm == 1 else (bb if zm == 2 and d < 0 else l[base]) if top - bot > 0: z = dict(dir=d, top=top, bot=bot, mid=(top + bot) / 2.0, t0=base, born=i, state=FRESH, end=-1, retests=0, inside=False) zi = len(zones) zones.append(z) live.append(z) side = 0 if d > 0 else 1 active[side] += 1 if p["max_live"] > 0 and active[side] > p["max_live"]: for k, y in enumerate(live): if (0 if y["dir"] > 0 else 1) == side: y["state"], y["end"] = DROPPED, i del live[k] active[side] -= 1 break out["dem_top" if d > 0 else "sup_top"][i] = top out["dem_bot" if d > 0 else "sup_bot"][i] = bot out["dem_base" if d > 0 else "sup_base"][i] = i - base breaks.append(dict(dir=d, level=sw[s]["p"], frm=sw[s]["i"], at=i, zone=zi)) out["live_dem"][i], out["live_sup"][i] = active[0], active[1] q = i - S if q >= S: for side in (1, -1): if is_swing(q, side): sw[0 if side > 0 else 1] = dict(i=q, p=h[q] if side > 0 else l[q], broken=False) if on_bar is not None: on_bar(i, zones) return out, zones, breaks, live, atr