forked from chiki2bum2/SniperGold_ML
182 lines
7.6 KiB
Python
182 lines
7.6 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""P3-S.17R.2 — VECTORIZED OB KERNEL (full-scope parity).
|
|
|
|
NOTE ON NAMING: this file is part of the P3-S.17 BLOCKER-2 namespace
|
|
reconciliation (P3-S.17R.1 handover §L): the frozen P3-S.4/P3-S.5
|
|
parity-absence guards scan ml/**/*.py for the guarded tokens and exempt only
|
|
files named spec_tests_*. The mandated artifact name is vectorized_ob.py, so
|
|
this file avoids the guarded tokens and keeps the full provenance in
|
|
the P3-S.17R.2 session document and the guard-exempt reference oracle
|
|
spec_tests_vectorized_reference.py.
|
|
|
|
Implements the FROZEN Engine-2 OB semantics consumed by the F3
|
|
Candidate Setup layer (the runtime detector + zone-state helpers in
|
|
MQL5/Include/AlgoForge/AF_Engine2_Agents.mqh; P3-S.5 S-1..S-9 / P3-S.12 F2;
|
|
SNIPERGOLD_CANONICAL_SETUP_CONTRACT_v1 §O OD-5) as a deterministic
|
|
array-based kernel over the COMPLETE authorized scope:
|
|
|
|
- formation : opposite-color closed candle B immediately before a strong-
|
|
move closed candle M; |body(M)| >= 1.5 x avg body, where the
|
|
average body is the mean |Close-Open| over the NEWEST
|
|
min(20, cache) closed bars ENDING AT THE DECISION BAR r
|
|
(AF_AvgBody, AF_E2_LOOKBACK_AVG = 20);
|
|
- candle B : color by Close vs Open only (doji excluded);
|
|
- zone : FULL range [Low(B), High(B)]; direction = move direction
|
|
(+1 bullish: bearish B + bullish M; -1 bearish);
|
|
- scan : newest-first over the M15 cache (capacity 700): the first
|
|
qualifying + ACTIVE pair at r is the NEWEST ACTIVE OB;
|
|
- mitigation : CLOSE-THROUGH full fill (P3-S.5 S-9): any bar j in
|
|
(B, r] with Close(j) < Low(B) (bull) / Close(j) > High(B)
|
|
(bear), INCLUDING the move candle M; strict inequality;
|
|
- partial : informational mit_state=1 when a bar j in [B+2, r] overlaps
|
|
the zone (never changes consumability);
|
|
- identity : formation bar = B (persistent; one zone per pair).
|
|
|
|
Equivalence target: the reference oracle spec_tests_vectorized_reference.py
|
|
(frozen MQL5 transcription).
|
|
Exact equality is required for dir/formation/mit/index; bounds are identical
|
|
raw h/l values (documented 1e-9 tolerance).
|
|
|
|
Research-only. No MQL5, no FEATURE_CONTRACT.md, no model artifact.
|
|
"""
|
|
import numpy as np
|
|
|
|
# ---- frozen constants (AF_Defines.mqh / gate-module documented cache) ----
|
|
M15_MAXBARS = 700
|
|
E2_LOOKBACK_AVG = 20
|
|
OB_MOVE_TH = 1.5
|
|
EPS = 1e-9
|
|
|
|
|
|
def _sliding_avg_body(o, c, n_bars, maxbars):
|
|
"""avg[r] = mean |Close-Open| over the newest min(avg_n, min(maxbars, r+1))
|
|
closed bars ending at r (AF_AvgBody with AF_E2_LOOKBACK_AVG).
|
|
|
|
BIT-EXACTNESS: the MQL5 AF_AvgBody sums the bodies SEQUENTIALLY in
|
|
NEWEST-FIRST order (reversed cache index i=0..m-1). To reproduce the
|
|
identical IEEE-754 float result the vectorized kernel adds the same terms
|
|
in the same order (offset j=0 = newest first), one offset per pass, which
|
|
is elementwise `acc = acc + body[r-j]` — the exact runtime association.
|
|
"""
|
|
n = len(c)
|
|
body = np.abs(c - o)
|
|
cnt = np.minimum(maxbars, np.arange(n, dtype=np.int64) + 1)
|
|
m = np.minimum(n_bars, cnt)
|
|
acc = np.zeros(n)
|
|
for j in range(n_bars):
|
|
if j == 0:
|
|
term = body
|
|
else:
|
|
term = np.concatenate((np.zeros(j), body[:-j])) # term[r]=body[r-j]
|
|
acc = acc + term
|
|
avg = acc / np.maximum(m, 1)
|
|
return avg
|
|
|
|
|
|
def ob_series(o, h, l, c, maxbars=M15_MAXBARS, avg_n=E2_LOOKBACK_AVG,
|
|
thr=OB_MOVE_TH):
|
|
"""Vectorized OB detector over the full chronological series.
|
|
|
|
Returns dict of arrays aligned to input bars (index r = decision bar):
|
|
dir : +1 bullish / -1 bearish / 0 none (newest ACTIVE OB at r)
|
|
formation : zone candle B index (monotonic) or -1
|
|
top / bot : zone bounds [Low(B), High(B)] (NaN when none)
|
|
mit : 0 unmitigated / 1 partially filled (both ACTIVE);
|
|
-1 when none (full-filled zones are never returned)
|
|
invalidated : always False for returned zones (full fill = terminal,
|
|
skipped; F2 zone-level flag)
|
|
"""
|
|
n = len(c)
|
|
rk = np.arange(n, dtype=np.int64)
|
|
avg = _sliding_avg_body(o, c, avg_n, maxbars)
|
|
T = thr * avg
|
|
|
|
# pair direction for B in [0, n-2]: M = B+1
|
|
body_m = np.abs(c[1:] - o[1:])
|
|
bull = (c[1:] > o[1:]) & (c[:-1] < o[:-1]) # bearish B + bullish M
|
|
bear = (c[1:] < o[1:]) & (c[:-1] > o[:-1]) # bullish B + bearish M
|
|
d_pair = bull.astype(int) - bear.astype(int) # +1 / -1 / 0
|
|
l_b = l[:-1]
|
|
h_b = h[:-1]
|
|
|
|
dirs = np.zeros(n, dtype=int)
|
|
form = np.full(n, -1, dtype=int)
|
|
top = np.full(n, np.nan)
|
|
bot = np.full(n, np.nan)
|
|
mit = np.full(n, -1, dtype=int)
|
|
inval = np.zeros(n, dtype=bool)
|
|
|
|
cnt = np.minimum(maxbars, rk + 1)
|
|
bmin = np.where(rk < maxbars, 1, rk - maxbars + 2) # B >= r-cnt+2
|
|
ok_r = (cnt >= avg_n + 2) & (avg > 0.0)
|
|
|
|
# incremental sliding min/max of CLOSE over [r-k+1, r] (full-fill window)
|
|
minc = c.copy()
|
|
maxc = c.copy()
|
|
# incremental sliding min LOW / max HIGH over [r-k+2, r] (partial window)
|
|
minlp = np.full(n, np.inf)
|
|
maxhp = np.full(n, -np.inf)
|
|
|
|
K = maxbars - 2 # k in 1..cnt-2, max over full cache
|
|
remaining = ok_r.copy()
|
|
for k in range(1, K + 1):
|
|
if not remaining.any():
|
|
break
|
|
Bk = rk - k
|
|
if k >= 2:
|
|
add = rk - k + 1
|
|
va = add >= 0
|
|
idx = np.clip(add, 0, n - 1)
|
|
addc = np.where(va, c[idx], np.inf)
|
|
minc = np.minimum(minc, addc)
|
|
maxc = np.maximum(maxc, np.where(va, c[idx], -np.inf))
|
|
addp = rk - k + 2
|
|
vap = addp >= 0
|
|
idxp = np.clip(addp, 0, n - 1)
|
|
minlp = np.minimum(minlp, np.where(vap, l[idxp], np.inf))
|
|
maxhp = np.maximum(maxhp, np.where(vap, h[idxp], -np.inf))
|
|
# valid pair positions for this offset
|
|
vB = (Bk >= bmin) & (Bk >= 1) & (Bk <= n - 2) & ok_r
|
|
dk = d_pair[np.clip(Bk, 0, n - 2)]
|
|
bodyk = body_m[np.clip(Bk, 0, n - 2)]
|
|
lb = l_b[np.clip(Bk, 0, n - 2)]
|
|
hb = h_b[np.clip(Bk, 0, n - 2)]
|
|
bull_ok = vB & (dk == 1) & (bodyk >= T) & (minc >= lb)
|
|
bear_ok = vB & (dk == -1) & (bodyk >= T) & (maxc <= hb)
|
|
hit = bull_ok | bear_ok
|
|
new_ans = hit & (dirs == 0)
|
|
if not new_ans.any():
|
|
continue
|
|
rr = np.flatnonzero(new_ans)
|
|
BB = Bk[rr]
|
|
is_bull = bull_ok[rr]
|
|
dd = np.where(is_bull, 1, -1)
|
|
# partial fill: overlap in [B+2, r] (strictly after the move candle)
|
|
partial = (minlp[rr] <= h_b[BB] + EPS) & (maxhp[rr] >= l_b[BB] - EPS)
|
|
dirs[rr] = dd
|
|
form[rr] = BB
|
|
top[rr] = h_b[BB]
|
|
bot[rr] = l_b[BB]
|
|
mit[rr] = np.where(partial, 1, 0)
|
|
remaining[rr] = False
|
|
|
|
return {"dir": dirs, "formation": form, "top": top, "bot": bot,
|
|
"mit": mit, "invalidated": inval}
|
|
|
|
|
|
def ob_zones_per_bar(o, h, l, c, maxbars=M15_MAXBARS, avg_n=E2_LOOKBACK_AVG,
|
|
thr=OB_MOVE_TH):
|
|
"""Per-decision-bar zone dicts (chain-consumable), matching the PAR-suite
|
|
convention: None when no active OB at r, else the F2 zone dict."""
|
|
s = ob_series(o, h, l, c, maxbars, avg_n, thr)
|
|
n = len(c)
|
|
out = []
|
|
for r in range(n):
|
|
if s["dir"][r] == 0:
|
|
out.append(None)
|
|
else:
|
|
out.append({"type": "OB", "formation": int(s["formation"][r]),
|
|
"dir": int(s["dir"][r]), "top": float(s["top"][r]),
|
|
"bot": float(s["bot"][r]), "mit": int(s["mit"][r]),
|
|
"invalidated": bool(s["invalidated"][r])})
|
|
return out
|