forked from chiki2bum2/SniperGold_ML
215 lines
9.8 KiB
Python
215 lines
9.8 KiB
Python
 | # -*- coding: utf-8 -*-
| ||
"""Liquidity Sweep EVENT — canonical research port (P3-S.2 / P3-S.11 frozen).
| |||
| |||
P3-S.17R.1 — Task A. This module REPLACES the legacy persistent-latch proxy
| |||
`smc_semantic_common.sweep_state` with a faithful *semantic EVENT* port of the
| |||
frozen runtime detector + the frozen F1 event contract. It exists so the
| |||
research-side Candidate Setup extractor can consume the SAME sweep semantics as
| |||
the frozen MQL5 runtime (AF_Engine2_Agents / AlgoForge_Backtest_Baseline.mq5
| |||
ProcessStructure + DetectLiquidityGrabs) instead of a state latch without expiry.
| |||
| |||
SOURCE OF TRUTH (frozen, unmodified):
| |||
- docs/SMC_LIQUIDITY_SWEEP_SPEC_v1.md (P3-S.2 project semantic spec)
| |||
- docs/SNIPERGOLD_CANONICAL_SETUP_CONTRACT_v1.md §K / §M (EVENT contract)
| |||
- docs/P3_S11_EVENT_CONTRACT_REPAIR.md (F1 EVENT {ts,dir,valid_until,
| |||
superseded}
| |||
- MQL5 runtime: AlgoForge_Backtest_Baseline.mq5
| |||
ProcessStructure(len=5,internal=true) (internal pivot emission)
| |||
DetectLiquidityGrabs(...) (sweep detection geometry)
| |||
g_feat[7] + AFEventSweep (P3-S.0 / F1) (event lifecycle/validity)
| |||
| |||
THE CONVENTION (what "Liquidity Sweep" means HERE — an EVENT, not a latch):
| |||
- A Liquidity Sweep is an EVENT with 4 fields: onset bar, direction, a bounded
| |||
validity window W_sweep, and a superseded flag.
| |||
- It is NOT "the latest sweep forever". Expiry = W_sweep bars after onset.
| |||
- Supersession: a NEWER onset replaces (refreshes) the active event.
| |||
| |||
DETECTION GEOMETRY (byte-locked to the runtime, NOT a new formula):
| |||
1. Internal swing pivot at bar p (fractal length 5 one-sided, i.e. 5 left + 5
| |||
right confirmations, strict inequality, equal bars do NOT cancel):
| |||
begin = max(startBar, 2*5); for i in [begin, total): p = i-5; p>=5
| |||
IsPivotHigh: idx-5>=0 && idx+5<total && all h[p]>h[p± k], k=1..5
| |||
IsPivotLow : idx-5>=0 && idx+5<total && all l[p]<l[p± k], k=1..5
| |||
2. For each confirmed internal pivot, scan b in (p, min(total-1, p+8)]:
| |||
buy-side swept : isHigh && high[b]>lvl && close[b]<lvl
| |||
-> dir=-1 (bearish intent)
| |||
sell-side swept: !isHigh && low[b]<lvl && close[b]>lvl
| |||
-> dir=+1 (bullish intent)
| |||
first matching bar per pivot sets an onset; `if b>cur_onset` (strictly
| |||
newer bar wins) matches the runtime tie-break. Later bars that satisfy
| |||
the condition in the same window are NOT new events for that pivot.
| |||
3. EVENT materialization (F1): each onset is an EVENT with
| |||
valid_until = onset + W_sweep (40); a newer onset supersedes the old.
| |||
| |||
DIRECTION (SPEC §5, not variable-name):
| |||
buy-side liquidity taken -> -1 (BEARISH); sell-side taken -> +1 (BULLISH).
| |||
| |||
CONSUMERS MUST apply the validity window: active iff onset <= r <= onset+W_sweep.
| |||
| |||
The legacy `smc_semantic_common.sweep_state` (a persistent latch marking every
| |||
bar >= onset as active) is NOT this module's semantics, and is NOT used by the
| |||
Candidate-Setup extractor path. It remains untouched inside the frozen P3-S.2
| |||
diagnostic suites which intentionally MEASURED that legacy state.
| |||
| |||
This file is research-only. No MQL5, no FEATURE_CONTRACT, no model artifact.
| |||
"""
| |||
import numpy as np
| |||
| |||
# Frozen constants (P3-S.2 §M / AF_Defines.mqh / AF_BT_*)
| |||
INTERNAL_LEN = 5 # internal pivot fractal one-sided length
| |||
GRAB_WINDOW = 8 # AF_E2_SWEEP_LOOKBACK / AF_BT_GRAB_WIN probe window
| |||
W_SWEEP = 40 # F1 sweep EVENT validity (M15 bars), InpSeqWindow origin
| |||
| |||
| |||
def _is_pivot_high(h, idx, length, total):
| |||
"""Mirror IsPivotHigh (strictly higher than 5 bars each side, equal cancels)."""
| |||
if idx - length < 0 or idx + length >= total:
| |||
return False
| |||
v = h[idx]
| |||
for k in range(1, length + 1):
| |||
if h[idx - k] > v or h[idx + k] > v:
| |||
return False
| |||
return True
| |||
| |||
| |||
def _is_pivot_low(l, idx, length, total):
| |||
"""Mirror IsPivotLow (strictly lower than 5 bars each side)."""
| |||
if idx - length < 0 or idx + length >= total:
| |||
return False
| |||
v = l[idx]
| |||
for k in range(1, length + 1):
| |||
if l[idx - k] < v or l[idx + k] < v:
| |||
return False
| |||
return True
| |||
| |||
| |||
def internal_pivots(h, l, start_bar=None, length=INTERNAL_LEN):
| |||
"""Confirmed internal swing pivots, monotonic chronological index.
| |||
| |||
Mirrors ProcessStructure(len=length, internal=true): begin = max(start_bar,
| |||
2*length); for i in [begin, n): p = i-length; if p>=length and IsPivot(p):
| |||
append (p, level, is_high).
| |||
A pivot at p is CONFIRMED once the p+length confirmation bar is available.
| |||
start_bar : the runtime warm-up / analysis-start bar. The runtime uses a
| |||
large cache (AF_BT_MIN_BARS / begin=100); default None -> 100 to match
| |||
the real-data runtime warm-up. Synthetic geometry tests may pass a
| |||
smaller start_bar (e.g. 0 -> begin = 2*length).
| |||
Returns list of (p, level: float, is_high: bool) in ascending p order.
| |||
"""
| |||
n = len(h)
| |||
begin = 100 if start_bar is None else start_bar
| |||
begin = max(begin, 2 * length)
| |||
pivots = []
| |||
for i in range(begin, n):
| |||
p = i - length
| |||
if p >= length:
| |||
if _is_pivot_high(h, p, length, n):
| |||
pivots.append((int(p), float(h[p]), True))
| |||
if _is_pivot_low(l, p, length, n):
| |||
pivots.append((int(p), float(l[p]), False))
| |||
return pivots
| |||
| |||
| |||
def sweep_scans(h, l, c, pivots, grab_window=GRAB_WINDOW, n_total=None,
| |||
length=INTERNAL_LEN, effective_window=True):
| |||
"""Raw sweep detection over a fixed full trace (DetectLiquidityGrabs +
| |||
the frozen SPEC's effective-window consequence, SMC_LIQUIDITY_SWEEP_SPEC §4).
| |||
| |||
For each confirmed internal pivot, the first bar b that satisfies
| |||
wick-penetration + same-bar close-back sets an onset. Per the frozen spec's
| |||
authoritative consequence note, the confirmation bars p+1 .. p+(length) would
| |||
cancel a pivot, so the mathematically possible sweep bars are
| |||
b in [p+length+1, min(total-1, p+grab_window)] (i.e. the effective window,
| |||
for length=5: b in [p+6, p+8]). When `effective_window` is False the raw EA
| |||
window (p+1 .. p+grab) is used for comparison.
| |||
| |||
Returns list of raw onsets {"onset","dir","level","pivot"} chronological
| |||
(dedup by onset bar, newest wins on a same-bar tie, per SPEC §12[3]).
| |||
"""
| |||
total = n_total if n_total is not None else len(c)
| |||
events = []
| |||
first_b_offset = (length + 1) if effective_window else 1
| |||
for (p, lvl, is_high) in pivots:
| |||
if p >= total:
| |||
continue
| |||
last = min(total - 1, p + grab_window)
| |||
first = p + first_b_offset
| |||
if first > last:
| |||
continue
| |||
for b in range(first, last + 1):
| |||
if b >= total:
| |||
break
| |||
if is_high and h[b] > lvl and c[b] < lvl:
| |||
events.append({"onset": int(b), "dir": -1,
| |||
"level": float(lvl), "pivot": int(p)})
| |||
break
| |||
if (not is_high) and l[b] < lvl and c[b] > lvl:
| |||
events.append({"onset": int(b), "dir": 1,
| |||
"level": float(lvl), "pivot": int(p)})
| |||
break
| |||
# EMISSION ORDER (byte-locked to the runtime): a sweep onset bar b is
| |||
# visible at decision r = b, so NEW events are the RECORD HIGHS of onsets
| |||
# in BAR order, not in pivot order. The runtime DetectLiquidityGrabs keeps
| |||
# g_swpBar only when b > g_swpBar (strict), and processes pivots ascending
| |||
# p, so a same-bar tie resolves to the EARLIEST pivot. Sorting by
| |||
# (onset, pivot) and taking strictly-increasing onsets reproduces the
| |||
# runtime's new-event sequence exactly (verified against the byte-locked
| |||
# sliding-window reference on the full authorized scope).
| |||
events.sort(key=lambda e: (e["onset"], e["pivot"]))
| |||
latest_onset = -1
| |||
out = []
| |||
for e in events:
| |||
if e["onset"] > latest_onset:
| |||
latest_onset = e["onset"]
| |||
out.append(e)
| |||
return out
| |||
| |||
| |||
def sweep_events(h, l, c, start_bar=None, w_sweep=W_SWEEP,
| |||
grab_window=GRAB_WINDOW, effective_window=True):
| |||
"""F1 sweep EVENT port: internal pivots + DetectLiquidityGrabs scan +
| |||
EVENT materialization with validity & supersession.
| |||
| |||
Returns a dict:
| |||
events : list of EVENT dicts in onset order
| |||
{"onset","dir","level","pivot","valid_until","superseded"}
| |||
active : np.int array y[r] = dir of the ACTIVE (latest, in-validity)
| |||
event at decision bar r, else 0. (per-bar consumption view)
| |||
legacy_latch_compare : demonstration that this is NOT the latch (each bar
| |||
only carries a dir within W_sweep of its own onset).
| |||
Callers needing the Candidate-Setup chain must consume `events` (or the
| |||
active[] view) and apply the validity window themselves (see the events'
| |||
valid_until).
| |||
"""
| |||
pivots = internal_pivots(h, l, start_bar=start_bar)
| |||
raw = sweep_scans(h, l, c, pivots, grab_window=grab_window,
| |||
effective_window=effective_window)
| |||
| |||
n = len(c)
| |||
active = np.zeros(n, dtype=int)
| |||
# per-bar active dir = newest onset visible & in-validity (SPEC §9 lifecycle)
| |||
events = []
| |||
for e in raw:
| |||
on, d = e["onset"], e["dir"]
| |||
upto = min(n - 1, on + w_sweep)
| |||
# mark this window (later onset overwrites -> supersession by construction)
| |||
active[on:upto + 1] = d
| |||
ev = dict(e)
| |||
ev["valid_until"] = upto
| |||
ev["superseded"] = False
| |||
events.append(ev)
| |||
| |||
# supersession flags: a later onset replaces an earlier one permanently
| |||
for i in range(len(events)):
| |||
for j in range(i + 1, len(events)):
| |||
if events[j]["onset"] > events[i]["onset"]:
| |||
events[i]["superseded"] = True
| |||
return {
| |||
"events": events,
| |||
"active": active,
| |||
"pivots": pivots,
| |||
"n_pivots": len(pivots),
| |||
"n_onsets": len(events),
| |||
"w_sweep": int(w_sweep),
| |||
"grab_window": int(grab_window),
| |||
}
|