194 lines
9.2 KiB
Python
 | # SPDX-License-Identifier: LicenseRef-PolyForm-Noncommercial-1.0.0 (see LICENSE)
| ||
"""Independent Python reference for SMC Zones, bar by bar. Swings, sweeps, BOS/CHoCH, order block and FVG
| |||
written as a closed-bar state machine.
| |||
| |||
Everything at bar i is decided from bars <= i only; the last bar of the array is the forming bar and is never used.
| |||
A zone's creation bar and mitigation bar, once set, never change.
| |||
| |||
Rules (dir +1 = bullish, -1 = bearish):
| |||
ATR Wilder (RMA) of the true range, seeded with the SMA of the first `atr` values.
| |||
swing bar p is a swing high (low) if each of the `swing` bars to its left has a strictly lower high (higher low)
| |||
and each of the `swing` bars to its right a high not above (low not below). Known at bar p + swing.
| |||
BOS/CHoCH a close beyond the newest unbroken swing high (low). CHoCH when the previous break went the other way,
| |||
otherwise BOS. The first break ever is a BOS.
| |||
order block the zone left by a break: org = lowest low (highest high) bar from the broken swing to the break bar.
| |||
If org is a candle against the move (bearish candle for a bullish break), the OB candle is the last of the
| |||
run of against-candles starting at org, else the nearest against-candle among the 3 bars before org, else org.
| |||
Bullish zone: top = high[ob], bot = min(low[ob], low[org]); bearish: top = max(high[ob], high[org]),
| |||
bot = low[ob]. Drawn from the earlier of ob/org.
| |||
FVG bars i-2, i, gap = low[i] - high[i-2] (bullish) or low[i-2] - high[i] (bearish) >= fvg_min * ATR[i].
| |||
sweep a bar wicks beyond an unswept swing (confirmed, at most `sweep_look` bars old) by >= sweep_min * ATR and
| |||
closes back inside: the lowest low (highest high) such swing is reported. Every swing wicked through is
| |||
marked taken whether or not the bar closed back inside.
| |||
mitigation (bars after the zone's creation bar), modes: 0 touch (wick reaches the near edge), 1 midpoint (wick reaches
| |||
the middle), 2 far-edge wick, 3 far-edge close. Midpoint is inclusive within a relative 1e-10.
| |||
touch a bar whose wick reaches the near edge of a live zone that was not inside on the bar before.
| |||
session optional: zones and sweeps are only created on bars whose open hour (server time) is inside [start, end).
| |||
Per bar the order is: mitigation/touch of older zones, sweeps, FVG, break + order block, then the swing `swing` bars back."""
| |||
import numpy as np
| |||
| |||
NAN = np.nan
| |||
| |||
| |||
def true_range(h, l, c):
| |||
tr = h - l
| |||
tr[1:] = np.maximum(tr[1:], np.maximum(np.abs(h[1:] - c[:-1]), np.abs(l[1:] - c[:-1])))
| |||
return tr
| |||
| |||
| |||
def rma(x, n):
| |||
out = np.full(len(x), NAN)
| |||
if len(x) < n:
| |||
return out
| |||
out[n - 1] = x[:n].mean()
| |||
for i in range(n, len(x)):
| |||
out[i] = (out[i - 1] * (n - 1) + x[i]) / n
| |||
return out
| |||
| |||
| |||
def in_session(t, start, end):
| |||
hr = (int(t) // 3600) % 24
| |||
if start == end:
| |||
return True
| |||
return start <= hr < end if start < end else (hr >= start or hr < end)
| |||
| |||
| |||
def mitigated(z, mode, o, h, l, c):
| |||
"""Does this bar mitigate zone z?"""
| |||
top, bot = z["top"], z["bot"]
| |||
mid = (top + bot) / 2.0
| |||
tol = abs(mid) * 1e-10 # a wick exactly on the midpoint counts; (top+bot)/2 can be 1 ulp off in decimals
| |||
if z["dir"] > 0:
| |||
return {0: l <= top, 1: l <= mid + tol, 2: l <= bot, 3: c < bot}[mode]
| |||
return {0: h >= bot, 1: h >= mid - tol, 2: h >= top, 3: c > top}[mode]
| |||
| |||
| |||
def smc_zones(t, o, h, l, c, swing=3, atr=14, fvg_min=0.20, ob_mit=3, fvg_mit=2, sweep_look=60, sweep_min=0.10,
| |||
use_session=False, sess_start=7, sess_end=17):
| |||
n = len(c)
| |||
A = rma(true_range(h, l, c), atr)
| |||
nan = lambda: np.full(n, NAN)
| |||
out = {k: nan() for k in ("ob_bull_top", "ob_bull_bot", "ob_bear_top", "ob_bear_bot", "fvg_bull_top", "fvg_bull_bot",
| |||
"fvg_bear_top", "fvg_bear_bot", "struct", "level", "sweep_bull", "sweep_bear",
| |||
"mitigated", "touched", "active_bull", "active_bear")}
| |||
zones, swings, breaks, sweeps, live = [], [], [], [], []
| |||
newest = {1: None, -1: None} # newest swing per side (side +1 = high)
| |||
trend = 0
| |||
swept_at = {1: -1, -1: -1}
| |||
| |||
def is_swing(p, side):
| |||
for k in range(1, swing + 1):
| |||
if side > 0 and (h[p - k] >= h[p] or h[p + k] > h[p]):
| |||
return False
| |||
if side < 0 and (l[p - k] <= l[p] or l[p + k] < l[p]):
| |||
return False
| |||
return True
| |||
| |||
def far(i, d):
| |||
return l[i] if d > 0 else h[i]
| |||
| |||
def near(i, d):
| |||
return h[i] if d > 0 else l[i]
| |||
| |||
def against(i, d):
| |||
return c[i] < o[i] if d > 0 else c[i] > o[i]
| |||
| |||
for i in range(n - 1): # the last bar is still forming
| |||
ok_atr = not np.isnan(A[i]) and A[i] > 0
| |||
if ok_atr:
| |||
a = A[i]
| |||
sess = (not use_session) or in_session(t[i], sess_start, sess_end)
| |||
# 1. older zones: mitigated or touched
| |||
n_mit = n_touch = 0
| |||
for z in live:
| |||
if z["born"] >= i:
| |||
continue
| |||
mode = ob_mit if z["kind"] == "ob" else fvg_mit
| |||
if mitigated(z, mode, o[i], h[i], l[i], c[i]):
| |||
z["end"] = i
| |||
n_mit += 1
| |||
continue
| |||
d = z["dir"]
| |||
inside = (l[i] <= z["top"]) if d > 0 else (h[i] >= z["bot"])
| |||
if inside and not z["inside"]:
| |||
z["touches"] += 1
| |||
n_touch += 1
| |||
z["inside"] = inside
| |||
live[:] = [z for z in live if z["end"] is None]
| |||
out["mitigated"][i], out["touched"][i] = n_mit, n_touch
| |||
# 2. sweeps
| |||
for d in (1, -1):
| |||
side, best = -d, None
| |||
for s in reversed(swings):
| |||
if i - s["i"] > sweep_look:
| |||
break
| |||
if s["side"] != side or s["swept"]:
| |||
continue
| |||
if (far(i, d) - (s["p"] - d * sweep_min * a)) * d < 0:
| |||
s["swept"] = True
| |||
if (c[i] - s["p"]) * d > 0 and (best is None or s["p"] * d < best["p"] * d):
| |||
best = s
| |||
if best is not None:
| |||
swept_at[d] = i
| |||
if sess:
| |||
out["sweep_bull" if d > 0 else "sweep_bear"][i] = best["p"]
| |||
sweeps.append(dict(dir=d, level=best["p"], at=i, frm=best["i"]))
| |||
# 3. fair value gap
| |||
if i >= 2:
| |||
for d in (1, -1):
| |||
gap = (l[i] - h[i - 2]) if d > 0 else (l[i - 2] - h[i])
| |||
if gap >= fvg_min * a and sess:
| |||
top, bot = (l[i], h[i - 2]) if d > 0 else (l[i - 2], h[i])
| |||
zones.append(dict(kind="fvg", dir=d, top=top, bot=bot, t0=i - 1, born=i, end=None,
| |||
touches=0, inside=False))
| |||
live.append(zones[-1])
| |||
k = "fvg_bull" if d > 0 else "fvg_bear"
| |||
out[k + "_top"][i], out[k + "_bot"][i] = top, bot
| |||
# 4. structure break and order block
| |||
for d in (1, -1):
| |||
s = newest[d]
| |||
if s is None or s["broken"] or (c[i] - s["p"]) * d <= 0:
| |||
continue
| |||
s["broken"] = True
| |||
choch = trend == -d
| |||
trend = d
| |||
org = s["i"]
| |||
for j in range(s["i"], i + 1):
| |||
if (far(j, d) - far(org, d)) * d < 0:
| |||
org = j
| |||
ob = -1
| |||
if against(org, d):
| |||
ob = org
| |||
while ob + 1 < i and against(ob + 1, d):
| |||
ob += 1
| |||
else:
| |||
for j in range(org - 1, max(0, org - 3) - 1, -1):
| |||
if against(j, d):
| |||
ob = j
| |||
break
| |||
if ob < 0:
| |||
ob = org
| |||
if d > 0:
| |||
top, bot = h[ob], min(l[ob], l[org])
| |||
else:
| |||
top, bot = max(h[ob], h[org]), l[ob]
| |||
if sess:
| |||
zones.append(dict(kind="ob", dir=d, top=top, bot=bot, t0=min(ob, org), born=i, end=None,
| |||
touches=0, inside=False))
| |||
live.append(zones[-1])
| |||
k = "ob_bull" if d > 0 else "ob_bear"
| |||
out[k + "_top"][i], out[k + "_bot"][i] = top, bot
| |||
out["struct"][i] = d * (2 if choch else 1)
| |||
out["level"][i] = s["p"]
| |||
breaks.append(dict(dir=d, choch=choch, level=s["p"], frm=s["i"], at=i))
| |||
out["active_bull"][i] = sum(1 for z in live if z["dir"] > 0)
| |||
out["active_bear"][i] = sum(1 for z in live if z["dir"] < 0)
| |||
# 5. the swing `swing` bars back is known from the next bar on
| |||
p = i - swing
| |||
if p >= swing:
| |||
for side in (1, -1):
| |||
if is_swing(p, side):
| |||
s = dict(i=p, p=h[p] if side > 0 else l[p], side=side, swept=False, broken=False)
| |||
swings.append(s)
| |||
newest[side] = s
| |||
return out, zones, breaks, sweeps
|