forked from chiki2bum2/SniperGold_ML
162 lines
6.9 KiB
Python
162 lines
6.9 KiB
Python
# -*- 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]
|