143 lines
7.2 KiB
Python
 | # 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
|