Aliwkarimi98/SniperGold_ML
Watch
1
0
Fork
You've already forked SniperGold_ML
 1
forked from chiki2bum2/SniperGold_ML
Code Pull requests Activity
SniperGold_ML/ml/p3/smc_semantic/spec_tests_engine2_gates.py

728 lines
27 KiB
Python
Raw Permalink Blame History

# -*- coding: utf-8 -*-
"""P3-S.17R.1 — ENGINE-2 GATE SERIES — faithful research port of the frozen
runtime H4/M30/M15 gate producers.
WHAT THIS MODULE IS
-------------------
The F3 Candidate Setup layer (AF_Engine2_Setup.mqh) does NOT compute its own
gate values: it CONSUMES per-M15-decision-bar semantic inputs
h4 : H4 context STATE direction (+1/-1/0)
m30 : M30 context STATE direction (+1/-1/0)
m15 : M15 entry CONDITION direction (+1/-1/0)
These inputs are produced by the frozen Engine-2 AGENTS (AF_Engine2_Agents.mqh):
Narrative agent on the H4 slot (AF_E2_TF_S1 = PERIOD_H4)
Context agent on the M30 slot (AF_E2_TF_S2 = PERIOD_M30)
Entry agent on the M15 slot (AF_E2_TF_S3 = PERIOD_M15)
Each agent is STATELESS per closed bar (P3-S.7 S-ST): its output is a pure
function of the Engine-1 closed-bar cache ending at the decision bar, and its
`dir` = +1 if fuzzy bias > AF_E2_DIR_TOL, -1 if bias < -AF_E2_DIR_TOL, else 0.
This module is a FAITHFUL Python port of those frozen rules (same constants,
same fuzzy membership functions, same rule weights, same tie-breaks, same
cache capacities). It exists because NO runnable MQL5 file drives the agents
or the F3 layer on the authorized historical scope (verified: the baseline EA
AlgoForge_Backtest_Baseline.mq5 is the F1/feature path only; AFAgent* and
AFSetupEngine are never invoked). The research side therefore reproduces the
frozen E-agent rule directly, per P3-S.17R.1 §14 ("replicate that meaning, not
merely the variable name") and §16 (port only what is needed; if a component is
missing, document the missing contract — see AS-OF note below).
AS-OF / TEMPORAL CONTRACT (P3-S.17R.1 §15; canonical §L; P3-S.7 S-T)
--------------------------------------------------------------------
For an M15 decision bar with open time t (close time tc = t + 900 s):
H4_asof(t) : newest CLOSED H4 bar with close_time <= tc
M30_asof(t) : newest CLOSED M30 bar with close_time <= tc
M15_asof(t) : the decision bar t itself (its own closed bar)
The bound is the DECISION-BAR CLOSE time tc, mirroring the runtime Engine-1
cache reality: the EA processes the M15 bar t on the first tick at/after tc,
and the HTF slot cache at that moment contains every HTF bar with close_time
<= tc (the forming HTF bar is dropped by the closed-bar lock in
AFEngine1MTF::Build). Consequently an HTF bar that closes EXACTLY at tc is
visible to that decision ("exactly at HTF close" boundary). The canonical
contract §L text states "close_time <= t" (open timestamp); the runtime
behaviour is "close_time <= tc" (close timestamp). The two differ by exactly
one M15 bar at HTF-close boundaries (M15 bars whose close coincides with an
HTF close). This module reproduces the RUNTIME behaviour (the parity
reference; §14/§15), and the boundary is pinned deterministically by GR-T11
(before / exactly at / after HTF close) and GR-T12 (no future HTF candle:
an HTF bar with close_time > tc is NEVER visible).
No future bar, no partial HTF candle, no hidden future state.
SIMPLIFICATIONS (documented, contract-safe)
-------------------------------------------
1. Zone partial-fill flag is NOT computed in the agents: AF_FVGZoneState /
AF_OBZoneState use mit_state = FULLY_MITIGATED + invalidated on FULL fill
only; IsActive() = NOT full-filled. Partial fill never changes IsActive(),
so the agent dir is a pure function of full-fill. (The F3 layer's zone
`mit` field is produced by the PAR-suite zone ports, not by this module.)
2. Full-fill scans are answered with a sliding-window min/max over the closed
cache ending at r (monotonic deque), which is EXACTLY equivalent to the
MQL5 backward loop bounded by the decision bar (no look-ahead).
3. M30 bars are resampled from the M15 feed (2 x M15 per M30), same underlying
ticks -> deterministic and identical to the runtime M30 OHLC.
Research-only module. No MQL5, no FEATURE_CONTRACT, no model artifact.
"""
import collections
import numpy as np
# ---- frozen Engine-2 constants (AF_Defines.mqh) ----------------------------
E2_MIN_BARS = 80 # AF_E2_MIN_BARS: minimum cache bars before a valid signal
E2_LOOKBACK_PD = 60 # AF_E2_LOOKBACK_PD: premium/discount & range window
E2_LOOKBACK_AVG = 20 # AF_E2_LOOKBACK_AVG: avg-body window (OB strong move)
E2_PIVOT_LOOKBACK = 200 # AF_E2_PIVOT_LOOKBACK: swing pivot search window
E2_SWEEP_LOOKBACK = 8 # AF_E2_SWEEP_LOOKBACK: E2 sweep scan window
E2_FVG_LOOKBACK = 40 # AF_E2_FVG_LOOKBACK: FVG search window (Context agent)
E2_DIR_TOL = 0.05 # AF_E2_DIR_TOL: agent dir threshold on fuzzy bias
E2_MAX_PIVOTS = 64 # AF_E2_MAX_PIVOTS: max stored pivots per side
DISP_TH = 1.6 # displacement: body >= 1.6 x avg body (AF_DetectDisplacement)
OB_MOVE_TH = 1.5 # AF_E3_MOVE_BODY: OB strong-move x avg body
# Cache capacities used by the baseline EA registration (AlgoForge_Backtest_
# Baseline.mq5 OnInit): M15 = InpMaxBars = 700; H1/H4/D1 = 250. M30 is not
# registered there; the F3 path would register it like the other HTF slots,
# so the research port uses 250 for M30 (documented assumption, bounded by all
# agent lookbacks <= 200).
M15_MAXBARS = 700
HTF_MAXBARS = 250
# ---------------------------------------------------------------------------
# Sliding-window min/max (monotonic deque) — exact equivalent of the bounded
# backward fill scans in AF_FVGZoneState / AF_OBZoneState at decision bar r.
# ---------------------------------------------------------------------------
class _WindowMM:
def __init__(self):
self._qmin = collections.deque()
self._qmax = collections.deque()
self._r = -1
def push(self, vmin, vmax):
self._r += 1
while self._qmin and self._qmin[-1][1] >= vmin:
self._qmin.pop()
self._qmin.append((self._r, vmin))
while self._qmax and self._qmax[-1][1] <= vmax:
self._qmax.pop()
self._qmax.append((self._r, vmax))
def query(self, a):
"""min over [a, r] and max over [a, r]; a <= r."""
while self._qmin and self._qmin[0][0] < a:
self._qmin.popleft()
while self._qmax and self._qmax[0][0] < a:
self._qmax.popleft()
mn = self._qmin[0][1] if self._qmin else None
mx = self._qmax[0][1] if self._qmax else None
return mn, mx
# ---------------------------------------------------------------------------
# Fuzzy membership functions (AF_Engine2_Agents.mqh)
# ---------------------------------------------------------------------------
def clamp01(v):
return 0.0 if v < 0.0 else (1.0 if v > 1.0 else v)
def mf_trap(x, a, b, c, d):
if x <= a or x >= d:
return 0.0
if x >= b and x <= c:
return 1.0
lo = (x - a) / (b - a) if b > a else 1.0
hi = (d - x) / (d - c) if d > c else 1.0
return lo if x < b else hi
def mf_tri(x, a, b, c):
if x <= a or x >= c:
return 0.0
if x == b:
return 1.0
lo = (x - a) / (b - a) if b > a else 1.0
hi = (c - x) / (c - b) if c > b else 1.0
return lo if x < b else hi
class _FuzzyEval:
"""AFFuzzyEval (Mamdani-light): weighted rule accumulation -> bias/dir."""
def __init__(self):
self.buy_acc = 0.0
self.sell_acc = 0.0
self.w_tot = 0.0
def rule(self, buy_side, fire, weight):
if fire <= 0.0 or weight <= 0.0:
return
self.w_tot += weight
if buy_side:
self.buy_acc += fire * weight
else:
self.sell_acc += fire * weight
def finalize(self):
denom = self.w_tot if self.w_tot > 0.0 else 1.0
buy = clamp01(self.buy_acc / denom)
sell = clamp01(self.sell_acc / denom)
bias = buy - sell
d = 1 if bias > E2_DIR_TOL else (-1 if bias < -E2_DIR_TOL else 0)
return {"buy": buy, "sell": sell, "bias": bias,
"confidence": max(buy, sell), "dir": d}
# ---------------------------------------------------------------------------
# Per-TF closed-bar primitives (reversed-index semantics of Engine-1 cache)
# All *_at(o,h,l,c, r, ...) functions evaluate AT closed chronological bar r
# using only bars [max(0, r-maxbars+1), r] (closed-bar lock, no look-ahead).
# ---------------------------------------------------------------------------
def _atr(o, h, l, c, r, period=14, maxbars=HTF_MAXBARS):
cnt = min(maxbars, r + 1)
if cnt <= 0 or period < 1:
return 0.0
n = min(period, cnt)
s = 0.0
for k in range(n):
b = r - k
tr = h[b] - l[b]
j = k + 1
if j < cnt:
pc = c[r - j]
t1 = abs(h[b] - pc)
t2 = abs(l[b] - pc)
tr = max(tr, t1, t2)
s += tr
return s / n
def _avg_body(o, h, l, c, r, n=20, maxbars=HTF_MAXBARS):
cnt = min(maxbars, r + 1)
m = min(n, cnt)
if m <= 0:
return 0.0
s = 0.0
for k in range(m):
b = r - k
s += abs(c[b] - o[b])
return s / m
def build_swing(o, h, l, c, r, lookback=E2_PIVOT_LOOKBACK, maxbars=HTF_MAXBARS):
"""AF_BuildSwing: fractal 2-left/2-right pivots on closed bars; returns
(highs, lows) lists of (chrono_bar, price) NEWEST-first (reversed order),
capped at AF_E2_MAX_PIVOTS per side."""
cnt = min(maxbars, r + 1)
highs = []
lows = []
if cnt < 5:
return highs, lows
max_idx = min(cnt - 3, lookback if lookback > 0 else cnt)
for k in range(2, max_idx + 1):
b = r - k
vh = h[b]
if (vh > h[b + 1] and vh > h[b + 2] and vh > h[b - 1] and vh > h[b - 2]):
if len(highs) < E2_MAX_PIVOTS:
highs.append((b, float(vh)))
vl = l[b]
if (vl < l[b + 1] and vl < l[b + 2] and vl < l[b - 1] and vl < l[b - 2]):
if len(lows) < E2_MAX_PIVOTS:
lows.append((b, float(vl)))
return highs, lows
def trend_from_swing(highs, lows):
"""AF_TrendFromSwing: +1/-1/0 from the newest up-to-4 pivot sequence."""
up = dn = pairs = 0
n_h = min(4, len(highs))
n_l = min(4, len(lows))
for i in range(n_h - 1):
if highs[i][1] > highs[i + 1][1]:
up += 1
else:
dn += 1
pairs += 1
for i in range(n_l - 1):
if lows[i][1] > lows[i + 1][1]:
up += 1
else:
dn += 1
pairs += 1
if pairs <= 0:
return 0, 0.0
clarity = float(max(up, dn)) / float(pairs)
trend = 1 if up > dn else (-1 if dn > up else 0)
return trend, clarity
def detect_choch(o, h, l, c, r, highs, lows, prev_trend):
"""AF_DetectChoch: close break of the newest swing pivot, prior-trend gated."""
if len(highs) < 1 or len(lows) < 1:
return 0
close = c[r]
if prev_trend < 0 and close > highs[0][1]:
return 1
if prev_trend > 0 and close < lows[0][1]:
return -1
return 0
def detect_bos(o, h, l, c, r, highs, lows, trend):
if len(highs) < 1 or len(lows) < 1:
return 0
close = c[r]
if trend > 0 and close > highs[0][1]:
return 1
if trend < 0 and close < lows[0][1]:
return -1
return 0
def detect_sweep_e2(o, h, l, c, r, highs, lows, lookback=E2_SWEEP_LOOKBACK,
maxbars=HTF_MAXBARS):
"""AF_DetectSweep (Engine-2 agent version): level = min/max of the 2 newest
lows/highs; newest-first scan for wick pen + same-bar close-back."""
if len(lows) < 1 or len(highs) < 1:
return 0
cnt = min(maxbars, r + 1)
n = min(lookback, cnt - 1)
if n < 1:
return 0
level_low = min(lows[0][1], lows[1][1]) if len(lows) >= 2 else lows[0][1]
level_high = max(highs[0][1], highs[1][1]) if len(highs) >= 2 else highs[0][1]
best_bull = -1
best_bear = -1
for k in range(n):
b = r - k
if best_bull < 0 and l[b] < level_low and c[b] > level_low:
best_bull = k
if best_bear < 0 and h[b] > level_high and c[b] < level_high:
best_bear = k
if best_bull >= 0 and (best_bear < 0 or best_bull <= best_bear):
return 1
if best_bear >= 0:
return -1
return 0
def detect_displacement(o, h, l, c, r, avg_n=E2_LOOKBACK_AVG, maxbars=HTF_MAXBARS):
avg = _avg_body(o, h, l, c, r, avg_n, maxbars)
if avg <= 0.0:
return 0
body = abs(c[r] - o[r])
if body < DISP_TH * avg:
return 0
return 1 if c[r] > o[r] else -1
def range_stat(o, h, l, c, r, lookback=E2_LOOKBACK_PD, maxbars=HTF_MAXBARS):
"""AF_RangeStat: (rmin, rmax, pos) over the newest `lookback` closed bars."""
cnt = min(maxbars, r + 1)
n = min(lookback, cnt)
if n < 2:
return 0.0, 0.0, 0.5
rmin = min(l[r - k] for k in range(n))
rmax = max(h[r - k] for k in range(n))
if rmax > rmin:
pos = clamp01((c[r] - rmin) / (rmax - rmin))
else:
pos = 0.5
return rmin, rmax, pos
def find_order_block(o, h, l, c, r, wm, avg_n=E2_LOOKBACK_AVG,
maxbars=HTF_MAXBARS):
"""AF_FindOrderBlock: newest qualifying OB (reversed scan i=1..cnt-2).
M at reversed i-1 (strong body >= 1.5 avg), B at reversed i (opposite
color), zone = full range of B, active (not close-through full-filled).
wm = _WindowMM over the CLOSED cache ending at r (for the fill query).
Returns (dir, top, bot) or (0, 0.0, 0.0)."""
cnt = min(maxbars, r + 1)
if cnt < avg_n + 2:
return 0, 0.0, 0.0
avg = _avg_body(o, h, l, c, r, avg_n, maxbars)
if avg <= 0.0:
return 0, 0.0, 0.0
for k in range(1, cnt - 1):
b_m = r - k + 1 # M candle (newer, reversed idx k-1)
b_b = r - k # B candle (older, reversed idx k)
body_prev = abs(c[b_m] - o[b_m])
if body_prev < OB_MOVE_TH * avg:
continue
up_move = c[b_m] > o[b_m]
d = 0
if up_move:
if c[b_b] < o[b_b]:
d = 1
else:
if c[b_b] > o[b_b]:
d = -1
if d == 0:
continue
top = h[b_b]
bot = l[b_b]
# full-fill (close-through): any closed bar newer than B beyond the zone
mn, mx = wm.query(b_b + 1)
if d > 0 and mn is not None and mn < bot:
continue
if d < 0 and mx is not None and mx > top:
continue
return d, float(top), float(bot)
return 0, 0.0, 0.0
def find_fvg(o, h, l, c, r, wm, lookback=E2_FVG_LOOKBACK, maxbars=HTF_MAXBARS):
"""AF_FindFVG: newest eligible C3 (reversed i=0..), wick geometry, active
(not wick full-filled). Returns (dir, top, bot) or (0,0,0)."""
cnt = min(maxbars, r + 1)
if cnt < 3:
return 0, 0.0, 0.0
max_idx = min(cnt - 3, lookback if lookback > 0 else cnt)
for k in range(max_idx + 1):
b_c3 = r - k # C3 (newest of the three)
b_c1 = r - k - 2 # C1 (oldest of the three)
l0 = l[b_c3]
h2 = h[b_c1]
if l0 > h2: # bullish FVG: Low(C3) > High(C1)
bot = h2
top = l0
mn, _ = wm.query(b_c3 + 1)
if mn is not None and mn <= bot:
continue # wick full-filled (bull: Low(j) <= bot)
return 1, float(top), float(bot)
h0 = h[b_c3]
l2 = l[b_c1]
if h0 < l2: # bearish FVG: High(C3) < Low(C1)
bot = h0
top = l2
_, mx = wm.query(b_c3 + 1)
if mx is not None and mx >= top:
continue # wick full-filled (bear: High(j) >= top)
return -1, float(top), float(bot)
return 0, 0.0, 0.0
def pattern_pa(o, h, l, c, r):
"""AF_PatternPA: engulfing / pin bar / inside bar / momentum / close-pos."""
if r < 3:
return 0, 0.0
o0, h0, l0, c0 = o[r], h[r], l[r], c[r]
o1, h1, l1, c1 = o[r - 1], h[r - 1], l[r - 1], c[r - 1]
b0 = c0 > o0
b1 = c1 > o1
body0 = abs(c0 - o0)
rng0 = h0 - l0
best = 0
best_str = 0.0
if b0 and not b1 and c0 >= o1 and o0 <= c1:
best, best_str = 1, 1.0
elif not b0 and b1 and c0 <= o1 and o0 >= c1:
best, best_str = -1, 1.0
if rng0 > 0.0:
lower_wick = min(o0, c0) - l0
upper_wick = h0 - max(o0, c0)
if body0 > 0.0 and lower_wick > 2.0 * body0 and upper_wick < 0.5 * body0 \
and best_str < 0.85:
best, best_str = 1, 0.85
if body0 > 0.0 and upper_wick > 2.0 * body0 and lower_wick < 0.5 * body0 \
and best_str < 0.85:
best, best_str = -1, 0.85
if h0 <= h1 + 1e-12 and l0 >= l1 - 1e-12:
if b1 and best_str < 0.6:
best, best_str = 1, 0.6
if not b1 and best_str < 0.6:
best, best_str = -1, 0.6
if b0 and b1 and best_str < 0.7:
best, best_str = 1, 0.7
if not b0 and not b1 and best_str < 0.7:
best, best_str = -1, 0.7
if rng0 > 0.0:
pos = (c0 - l0) / rng0
if pos > 0.70 and best_str < 0.5:
best, best_str = 1, 0.5
if pos < 0.30 and best_str < 0.5:
best, best_str = -1, 0.5
return best, best_str
# ---------------------------------------------------------------------------
# AGENTS — per closed TF bar r (chronological). dir = gate state value.
# ---------------------------------------------------------------------------
def narrative_agent(o, h, l, c, r, maxbars=HTF_MAXBARS):
"""AFAgentNarrative.Compute -> dir. Returns dict(dir, bias, reason_summary)."""
cnt = min(maxbars, r + 1)
if cnt < E2_MIN_BARS:
return {"dir": 0, "bias": 0.0, "reason": "data kurang"}
highs, lows = build_swing(o, h, l, c, r, E2_PIVOT_LOOKBACK, maxbars)
trend, clarity = trend_from_swing(highs, lows)
choch = detect_choch(o, h, l, c, r, highs, lows, trend)
bos = detect_bos(o, h, l, c, r, highs, lows, trend)
sweep = detect_sweep_e2(o, h, l, c, r, highs, lows, E2_SWEEP_LOOKBACK, maxbars)
_rmin, _rmax, pos = range_stat(o, h, l, c, r, E2_LOOKBACK_PD, maxbars)
m_bull = (0.35 + 0.65 * clarity) if trend > 0 else 0.0
m_bear = (0.35 + 0.65 * clarity) if trend < 0 else 0.0
m_disc = mf_trap(pos, 0.0, 0.0, 0.30, 0.45)
m_prem = mf_trap(pos, 0.55, 0.70, 1.0, 1.0)
w_struct = 0.40 * (0.5 + 0.5 * clarity)
w_liq = 0.35
w_zone = 0.25 * (1.5 - 0.5 * clarity)
w_sum = w_struct + w_liq + w_zone
w_struct /= w_sum
w_liq /= w_sum
w_zone /= w_sum
fz = _FuzzyEval()
fz.rule(True, m_bull, w_struct)
fz.rule(False, m_bear, w_struct)
fz.rule(True, 1.0 if sweep > 0 else 0.0, w_liq)
fz.rule(False, 1.0 if sweep < 0 else 0.0, w_liq)
fz.rule(True, m_disc, w_zone)
fz.rule(False, m_prem, w_zone)
if choch > 0:
fz.rule(True, 0.8, w_struct * 0.5)
if choch < 0:
fz.rule(False, 0.8, w_struct * 0.5)
if bos > 0 and trend > 0:
fz.rule(True, 0.7, w_struct * 0.3)
if bos < 0 and trend < 0:
fz.rule(False, 0.7, w_struct * 0.3)
out = fz.finalize()
out["reason"] = "Narrative@" + ("H4" if maxbars == HTF_MAXBARS else "TF")
return out
def context_agent(o, h, l, c, r, maxbars=HTF_MAXBARS):
"""AFAgentContext.Compute -> dir (Context / M30 gate)."""
cnt = min(maxbars, r + 1)
if cnt < E2_MIN_BARS:
return {"dir": 0, "bias": 0.0, "reason": "data kurang"}
close = c[r]
atr = _atr(o, h, l, c, r, 14, maxbars)
if atr <= 0.0:
return {"dir": 0, "bias": 0.0, "reason": "ATR 0"}
wm = _WindowMM()
for k in range(cnt):
b = r - k
wm.push(l[b], h[b])
ob_dir, ob_hi, ob_lo = find_order_block(o, h, l, c, r, wm, E2_LOOKBACK_AVG,
maxbars)
fv_dir, fv_hi, fv_lo = find_fvg(o, h, l, c, r, wm, E2_FVG_LOOKBACK, maxbars)
highs, lows = build_swing(o, h, l, c, r, E2_PIVOT_LOOKBACK, maxbars)
dist_sup = float("inf")
dist_res = float("inf")
for i in range(min(12, len(lows))):
d = close - lows[i][1]
if d >= 0.0 and d < dist_sup:
dist_sup = d
for i in range(min(12, len(highs))):
d = highs[i][1] - close
if d >= 0.0 and d < dist_res:
dist_res = d
m_at_sup = clamp01(1.0 - dist_sup / atr) if dist_sup < float("inf") else 0.0
m_at_res = clamp01(1.0 - dist_res / atr) if dist_res < float("inf") else 0.0
_rmin, _rmax, pos = range_stat(o, h, l, c, r, E2_LOOKBACK_PD, maxbars)
m_disc = mf_trap(pos, 0.0, 0.0, 0.30, 0.45)
m_prem = mf_trap(pos, 0.55, 0.70, 1.0, 1.0)
m_in_ob_bull = m_in_ob_bear = 0.0
m_in_fv_bull = m_in_fv_bear = 0.0
if ob_dir > 0:
m_in_ob_bull = mf_trap(close, ob_lo - 0.3 * atr, ob_lo, ob_hi, ob_hi + 0.3 * atr)
if ob_dir < 0:
m_in_ob_bear = mf_trap(close, ob_lo - 0.3 * atr, ob_lo, ob_hi, ob_hi + 0.3 * atr)
if fv_dir > 0:
m_in_fv_bull = mf_trap(close, fv_lo - 0.3 * atr, fv_lo, fv_hi, fv_hi + 0.3 * atr)
if fv_dir < 0:
m_in_fv_bear = mf_trap(close, fv_lo - 0.3 * atr, fv_lo, fv_hi, fv_hi + 0.3 * atr)
vol = atr / close if close != 0.0 else 0.0
vol_factor = 0.70 if vol > 0.002 else 1.0
w_ob, w_fv, w_sr, w_pd = 0.30, 0.25, 0.25 * vol_factor, 0.20 * vol_factor
w_sum2 = w_ob + w_fv + w_sr + w_pd
w_ob /= w_sum2
w_fv /= w_sum2
w_sr /= w_sum2
w_pd /= w_sum2
fz = _FuzzyEval()
fz.rule(True, m_in_ob_bull, w_ob)
fz.rule(False, m_in_ob_bear, w_ob)
fz.rule(True, m_in_fv_bull, w_fv)
fz.rule(False, m_in_fv_bear, w_fv)
fz.rule(True, m_at_sup, w_sr)
fz.rule(False, m_at_res, w_sr)
fz.rule(True, m_disc, w_pd)
fz.rule(False, m_prem, w_pd)
out = fz.finalize()
out["reason"] = "Context@" + ("M30" if maxbars == HTF_MAXBARS else "TF")
return out
def entry_agent(o, h, l, c, r, maxbars=M15_MAXBARS):
"""AFAgentEntry.Compute -> dir (Entry / M15 entry condition)."""
cnt = min(maxbars, r + 1)
if cnt < E2_MIN_BARS:
return {"dir": 0, "bias": 0.0, "reason": "data kurang"}
highs, lows = build_swing(o, h, l, c, r, E2_PIVOT_LOOKBACK, maxbars)
trend, _clarity = trend_from_swing(highs, lows)
sweep = detect_sweep_e2(o, h, l, c, r, highs, lows, E2_SWEEP_LOOKBACK, maxbars)
choch = detect_choch(o, h, l, c, r, highs, lows, trend)
disp = detect_displacement(o, h, l, c, r, E2_LOOKBACK_AVG, maxbars)
wm = _WindowMM()
for k in range(cnt):
b = r - k
wm.push(l[b], h[b])
ob_dir, ob_hi, ob_lo = find_order_block(o, h, l, c, r, wm, E2_LOOKBACK_AVG,
maxbars)
fv_dir, fv_hi, fv_lo = find_fvg(o, h, l, c, r, wm, 20, maxbars)
close = c[r]
atr = _atr(o, h, l, c, r, 14, maxbars)
if atr <= 0.0:
return {"dir": 0, "bias": 0.0, "reason": "ATR 0"}
m_sweep_bull = 1.0 if sweep > 0 else 0.0
m_sweep_bear = 1.0 if sweep < 0 else 0.0
m_choch_bull = 1.0 if choch > 0 else 0.0
m_choch_bear = 1.0 if choch < 0 else 0.0
m_disp_bull = 1.0 if disp > 0 else 0.0
m_disp_bear = 1.0 if disp < 0 else 0.0
m_zone_bull = m_zone_bear = 0.0
if ob_dir > 0:
m_zone_bull = max(m_zone_bull,
mf_trap(close, ob_lo - 0.3 * atr, ob_lo, ob_hi, ob_hi + 0.3 * atr))
if ob_dir < 0:
m_zone_bear = max(m_zone_bear,
mf_trap(close, ob_lo - 0.3 * atr, ob_lo, ob_hi, ob_hi + 0.3 * atr))
if fv_dir > 0:
m_zone_bull = max(m_zone_bull,
mf_trap(close, fv_lo - 0.3 * atr, fv_lo, fv_hi, fv_hi + 0.3 * atr))
if fv_dir < 0:
m_zone_bear = max(m_zone_bear,
mf_trap(close, fv_lo - 0.3 * atr, fv_lo, fv_hi, fv_hi + 0.3 * atr))
w_sweep, w_choch, w_disp, w_zone, w_setup = 0.25, 0.30, 0.20, 0.15, 0.30
if abs(disp) > 0:
w_disp *= 1.3
w_sum3 = w_sweep + w_choch + w_disp + w_zone + w_setup
w_sweep /= w_sum3
w_choch /= w_sum3
w_disp /= w_sum3
w_zone /= w_sum3
w_setup /= w_sum3
fz = _FuzzyEval()
fz.rule(True, m_sweep_bull, w_sweep)
fz.rule(False, m_sweep_bear, w_sweep)
fz.rule(True, m_choch_bull, w_choch)
fz.rule(False, m_choch_bear, w_choch)
fz.rule(True, m_disp_bull, w_disp)
fz.rule(False, m_disp_bear, w_disp)
fz.rule(True, m_zone_bull, w_zone)
fz.rule(False, m_zone_bear, w_zone)
conf_bull = max(m_sweep_bull, m_choch_bull)
conf_bear = max(m_sweep_bear, m_choch_bear)
fz.rule(True, min(m_zone_bull, conf_bull), w_setup)
fz.rule(False, min(m_zone_bear, conf_bear), w_setup)
out = fz.finalize()
out["reason"] = "Entry@M15"
return out
# ---------------------------------------------------------------------------
# SERIES + AS-OF GATE WIRING
# ---------------------------------------------------------------------------
def agent_series(agent_fn, o, h, l, c, maxbars):
"""dir for every closed bar of a TF (agent evaluated AT that bar's close)."""
n = len(c)
out = np.zeros(n, dtype=int)
for r in range(n):
out[r] = agent_fn(o, h, l, c, r, maxbars)["dir"]
return out
def m30_from_m15(t15, o15, h15, l15, c15):
"""Resample M15 -> M30 (2 x M15 per M30). Returns (t30, o30, h30, l30, c30).
t30 = M30 open time; close_time = t30 + 1800."""
n = len(t15) - (len(t15) % 2)
t30 = t15[0:n:2]
o30 = o15[0:n:2]
h30 = np.maximum(h15[0:n:2], h15[1:n:2])
l30 = np.minimum(l15[0:n:2], l15[1:n:2])
c30 = c15[1:n:2]
return t30, o30, h30, l30, c30
def as_of_index(htf_close, decision_close):
"""For each decision close time, the index of the last HTF bar with
close_time <= decision_close (runtime Engine-1 cache as-of)."""
return np.searchsorted(htf_close, decision_close, side="right") - 1
def gate_series(t15, o15, h15, l15, c15, h4, m30):
"""Build the full research H4/M30/M15 gate series aligned to M15 bars.
h4 = (t4, o4, h4a, l4a, c4a) H4 OHLC (chronological)
m30 = (t30, o30, h30a, l30a, c30a) M30 OHLC (chronological)
For M15 decision bar r (open t15[r], close tc = t15[r]+900):
h4_gate[r] = narrative dir of the newest H4 bar with close_time <= tc
m30_gate[r] = context dir of the newest M30 bar with close_time <= tc
m15_gate[r] = entry dir of the M15 bar r itself
Returns dict with h4/m30/m15 int arrays + provenance metadata.
"""
t4, o4, h4a, l4a, c4a = h4
t30, o30, h30a, l30a, c30a = m30
# agent dir per closed TF bar
h4_dir = agent_series(narrative_agent, o4, h4a, l4a, c4a, HTF_MAXBARS)
m30_dir = agent_series(context_agent, o30, h30a, l30a, c30a, HTF_MAXBARS)
m15_dir = agent_series(entry_agent, o15, h15, l15, c15, M15_MAXBARS)
# runtime as-of: newest closed HTF bar with close_time <= decision close
h4_close = t4 + 14400
m30_close = t30 + 1800
dc = t15 + 900
h4_idx = as_of_index(h4_close, dc)
m30_idx = as_of_index(m30_close, dc)
n = len(t15)
h4_gate = np.zeros(n, dtype=int)
m30_gate = np.zeros(n, dtype=int)
m15_gate = m15_dir.copy()
for r in range(n):
i4 = int(h4_idx[r])
if i4 >= 0:
h4_gate[r] = int(h4_dir[i4])
i30 = int(m30_idx[r])
if i30 >= 0:
m30_gate[r] = int(m30_dir[i30])
return {
"h4": h4_gate,
"m30": m30_gate,
"m15": m15_gate,
"h4_dir_per_h4_bar": h4_dir,
"m30_dir_per_m30_bar": m30_dir,
"m15_dir_per_m15_bar": m15_dir,
"as_of": "runtime Engine-1 cache: newest CLOSED HTF bar with "
"close_time <= decision-bar close (t+900); M15 = decision bar",
"constants": {
"E2_MIN_BARS": E2_MIN_BARS, "E2_DIR_TOL": E2_DIR_TOL,
"E2_PIVOT_LOOKBACK": E2_PIVOT_LOOKBACK,
"E2_SWEEP_LOOKBACK": E2_SWEEP_LOOKBACK,
"E2_FVG_LOOKBACK": E2_FVG_LOOKBACK,
"E2_LOOKBACK_PD": E2_LOOKBACK_PD,
"E2_LOOKBACK_AVG": E2_LOOKBACK_AVG,
"M15_MAXBARS": M15_MAXBARS, "HTF_MAXBARS": HTF_MAXBARS,
},
}
View git blame Copy permalink