SniperGold_ML/ml/p3/smc_semantic/sweep_event.py

# -*- 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),
    }