# -*- coding: utf-8 -*- """P3-S.17R.2 — VECTORIZED CHOCH KERNEL (full-scope runtime <-> research parity). Implements the FROZEN Engine-2 CHoCH semantics consumed by the F3 Candidate Setup layer (AF_BuildSwing + AF_TrendFromSwing + AF_DetectChoch / AF_DetectChochEvent, MQL5/Include/AlgoForge/AF_Engine2_Agents.mqh) as a deterministic, array-based kernel over the COMPLETE authorized historical scope, WITHOUT changing any frozen semantic: - pivot source : fractal 2-left/2-right, STRICT inequality on both sides (equal highs/lows do NOT form a pivot); - pivot window : per decision bar r, pivots on bars [max(2, r - min(min(maxbars,r+1)-3, lookback)), r-2] (lookback = E2_PIVOT_LOOKBACK = 200; cache cap 700); - trend : AF_TrendFromSwing over the NEWEST up-to-4 pivot highs and up-to-4 pivot lows (newest-first consecutive pairs; up = newer price > older price); - break condition : close-confirmed, STRICT: bullish close > newest pivot high requires prior trend < 0; bearish close < newest pivot low requires prior trend > 0; - event : per decision bar r (AF_DetectChochEvent onset = r), one dir per bar; closed-bar, no future visibility; - both sides : a CHoCH requires >= 1 pivot high AND >= 1 pivot low in the window (AF_DetectChoch early return). The 64-pivot-per-side cap (AF_E2_MAX_PIVOTS) only ever drops pivots OLDER than the newest 4 per side within the 200-bar window, so it cannot change the trend (newest-4) or the break (newest-1) — documented, not approximated. Equivalence target: vectorized_reference.ref_choch_* (the frozen MQL5 transcription). Exact equality is required for dir/trend/index; pivot levels are identical raw h/l values (compared with a 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_PIVOT_LOOKBACK = 200 E2_MAX_PIVOTS = 64 def choch_series(o, h, l, c, maxbars=M15_MAXBARS, lookback=E2_PIVOT_LOOKBACK): """Vectorized Engine-2 CHoCH over the full chronological series. Returns dict of arrays aligned to input bars (index r = decision bar): dir : +1 bullish / -1 bearish / 0 none at bar r (the EVENT dir) trend : prior internal trend used by the break precondition newest_high : level of the newest pivot high in the window (NaN none) newest_low : level of the newest pivot low in the window (NaN none) n_pivots_h/l : total pivot counts (whole series, diagnostic) events derived by the caller: onset r where dir[r] != 0. """ n = len(c) rk = np.arange(n, dtype=np.int64) dirs = np.zeros(n, dtype=int) trend = np.zeros(n, dtype=int) nh = np.full(n, np.nan) nl = np.full(n, np.nan) # ---- 1) pivot masks (strict fractal 2/2 on the full closed series) ---- ph = np.zeros(n, dtype=bool) pl = np.zeros(n, dtype=bool) if n >= 5: ph[2:n - 2] = ((h[2:n - 2] > h[0:n - 4]) & (h[2:n - 2] > h[1:n - 3]) & (h[2:n - 2] > h[3:n - 1]) & (h[2:n - 2] > h[4:n])) pl[2:n - 2] = ((l[2:n - 2] < l[0:n - 4]) & (l[2:n - 2] < l[1:n - 3]) & (l[2:n - 2] < l[3:n - 1]) & (l[2:n - 2] < l[4:n])) pos_h = np.flatnonzero(ph) pos_l = np.flatnonzero(pl) if n < 5 or pos_h.size == 0 or pos_l.size == 0: return {"dir": dirs, "trend": trend, "newest_high": nh, "newest_low": nl, "n_pivots_h": int(pos_h.size), "n_pivots_l": int(pos_l.size)} # ---- 2) per-decision-bar pivot window ---- cnt = np.minimum(maxbars, rk + 1) max_idx = np.minimum(cnt - 3, lookback) lower = np.maximum(2, rk - max_idx) # first pivot bar in window upper = rk - 2 # last pivot bar in window # ---- 3) newest pivot index per side (last pivot bar <= r-2) ---- ih = np.searchsorted(pos_h, upper, side="right") - 1 il = np.searchsorted(pos_l, upper, side="right") - 1 # first pivot index with bar >= lower fh = np.searchsorted(pos_h, lower, side="left") fl = np.searchsorted(pos_l, lower, side="left") # ---- 4) gather newest-4 levels per side (clipped; masked where invalid) ---- def gather(pos, idx): return h[pos[np.clip(idx, 0, pos.size - 1)]], \ l[pos[np.clip(idx, 0, pos.size - 1)]] ih0 = ih ih1 = ih - 1 ih2 = ih - 2 ih3 = ih - 3 il0 = il il1 = il - 1 il2 = il - 2 il3 = il - 3 hp0, _ = gather(pos_h, ih0) hp1, _ = gather(pos_h, ih1) hp2, _ = gather(pos_h, ih2) hp3, _ = gather(pos_h, ih3) _, lp0 = gather(pos_l, il0) _, lp1 = gather(pos_l, il1) _, lp2 = gather(pos_l, il2) _, lp3 = gather(pos_l, il3) valid_h = (ih >= fh) & (ih >= 0) valid_l = (il >= fl) & (il >= 0) # pair (i, i+1) valid iff the OLDER one (ih_{i+1}) is still >= fh m01h = ih1 >= fh m12h = ih2 >= fh m23h = ih3 >= fh m01l = il1 >= fl m12l = il2 >= fl m23l = il3 >= fl up = np.zeros(n, dtype=int) dn = np.zeros(n, dtype=int) up += (m01h & (hp0 > hp1)).astype(int) + (m12h & (hp1 > hp2)).astype(int) \ + (m23h & (hp2 > hp3)).astype(int) dn += (m01h & (hp0 <= hp1)).astype(int) + (m12h & (hp1 <= hp2)).astype(int) \ + (m23h & (hp2 <= hp3)).astype(int) up += (m01l & (lp0 > lp1)).astype(int) + (m12l & (lp1 > lp2)).astype(int) \ + (m23l & (lp2 > lp3)).astype(int) dn += (m01l & (lp0 <= lp1)).astype(int) + (m12l & (lp1 <= lp2)).astype(int) \ + (m23l & (lp2 <= lp3)).astype(int) trend = np.where(up > dn, 1, np.where(dn > up, -1, 0)) # ---- 5) newest levels + close-confirmed break ---- # AF_DetectChoch returns 0 when EITHER side has no pivot; the reference # mirrors that by reporting both levels as unavailable in that case. both = valid_h & valid_l nh = np.where(both, hp0, np.nan) nl = np.where(both, lp0, np.nan) close = c bull = both & (trend < 0) & (close > nh) bear = both & (trend > 0) & (close < nl) dirs = np.where(bull, 1, np.where(bear, -1, 0)) return {"dir": dirs, "trend": trend, "newest_high": nh, "newest_low": nl, "n_pivots_h": int(pos_h.size), "n_pivots_l": int(pos_l.size)} def choch_events(o, h, l, c, maxbars=M15_MAXBARS, lookback=E2_PIVOT_LOOKBACK): """CHoCH EVENT list derived from the vectorized kernel. Per the F1 EVENT contract (P3-S.11) + AF_DetectChochEvent, an onset is the decision bar r where the per-bar stateless detector returns dir != 0. Returns list of {"onset": r, "dir": dir} in chronological order. """ res = choch_series(o, h, l, c, maxbars, lookback) d = res["dir"] idx = np.flatnonzero(d != 0) return [{"onset": int(r), "dir": int(d[r])} for r in idx]