Warrior_EA/research/vol_target.py

"""
Volatility targets for the meta-label: Garman-Klass, and the 48-72h
expansion-vs-chop triple-barrier label.

Both estimators answer "how much range is coming", NOT "which way". That is the
whole point of the pivot: the direction question is what carries the neutral
majority-class trap, and this repo has already measured direction to be ~0 edge
out of sample.

(A) GARMAN-KLASS
----------------
For one bar with open O, high H, low L, close C, assuming zero drift and a
continuously-observed diffusion:

    sigma2_GK = 0.5 * (ln(H/L))^2 - (2*ln2 - 1) * (ln(C/O))^2

It is ~7.4x more efficient than the close-to-close estimator at equal sample
size, because the high and low carry path information that C alone discards.

THREE CAVEATS THAT MATTER ON CFD DATA, none of them cosmetic:

 1. GK IS NON-NEGATIVE FOR ANY VALID BAR -- do not "fix" it with a clamp.
    Since H >= max(O,C) and L <= min(O,C), ln(H/L) >= |ln(C/O)|, so
        GK >= (0.5 - (2ln2 - 1)) * ln(C/O)^2 = 0.1137 * ln(C/O)^2 >= 0.
    Verified numerically against the degenerate worst case (range == |C-O|):
    the observed minimum equals that algebraic floor to 1e-15.
    The REAL hazard is upstream: a corrupt bar (H < L, or an EMPTY_VALUE that
    reads as DBL_MAX) does NOT go negative -- it produces a plausible POSITIVE
    variance that no sanity check will catch. Guard the OHLC inputs, exactly as
    FeatureBuilder.mqh already guards close/high/low against EMPTY_VALUE, and
    never rely on a negativity test to detect bad data.
 2. GK IGNORES THE OVERNIGHT GAP. It assumes the bar opens where the last one
    closed. Index CFDs gap across the daily maintenance break and across the
    weekend, so on D1 GK systematically UNDERSTATES true variance. Use
    Yang-Zhang when the gap is part of the risk you are sizing against --
    which, for a multi-day intra-week hold, it is.
 3. DISCRETE SAMPLING BIASES H AND L INWARD. The observed high of a bar built
    from ticks is below the true continuous-path supremum, so GK is biased
    slightly low. Second-order next to (2) and can be ignored.

(B) THE 48-72h EXPANSION LABEL
------------------------------
Triple-barrier variant, UNSIGNED. From a trigger at t0:

    threshold  = K * ATR(t0)                      (K default 2.0)
    path       = bars t0+1 .. t0+H_max            (H_max = 72h)
    expansion  = max( max(High_path) - entry , entry - min(Low_path) )
    y          = 1 if expansion >= threshold else 0

y=1 is "variance expanded enough to pay for a multi-day hold"; y=0 is chop.
There is no third class and no sign, so there is no neutral bucket to collapse
into -- which is the specific failure mode of the current Argmax3 head.

THE WEEKEND INTERACTION IS NOT OPTIONAL. A Wednesday trigger plus 72 calendar
hours lands on Saturday. If the EA is flat by Friday afternoon by mandate, a
label that scores Saturday's range is scoring a position the strategy is
FORBIDDEN to hold. Every such row teaches a horizon that can never be traded.
So the path is clipped at the Friday liquidation timestamp, and a row whose
clipped path is shorter than H_min (48h) is DROPPED, not labelled 0 -- labelling
it 0 would manufacture a fake chop class that is really just "ran out of week",
and that class would correlate with day-of-week, which the network already sees
through its cyclical sin/cos time features. It would learn the calendar, not
the volatility.
"""

from __future__ import annotations

import numpy as np

LN2 = np.log(2.0)


# ----------------------------------------------------------------- estimators
def garman_klass_var(o, h, l, c) -> np.ndarray:
    """Per-bar GK variance. Non-negative for any valid OHLC bar (see caveat 1);
    a negative result therefore means CORRUPT INPUT, not an edge case."""
    o, h, l, c = (np.asarray(v, dtype=float) for v in (o, h, l, c))
    return 0.5 * np.log(h / l) ** 2 - (2.0 * LN2 - 1.0) * np.log(c / o) ** 2


def garman_klass_sigma(o, h, l, c, window: int) -> np.ndarray:
    """Rolling GK sigma over `window` bars (oldest -> newest), NaN warm-up.

    The max(...,0) before the root is defence against corrupt OHLC only -- it
    is unreachable on valid bars, and if it ever fires the data is at fault.
    """
    v = garman_klass_var(o, h, l, c)
    out = np.full(v.shape, np.nan)
    if window <= 0 or window > len(v):
        return out
    csum = np.concatenate([[0.0], np.nancumsum(v)])
    mean_var = (csum[window:] - csum[:-window]) / window
    out[window - 1:] = np.sqrt(np.maximum(mean_var, 0.0))
    return out


def parkinson_sigma(h, l, window: int) -> np.ndarray:
    """High-low only. No drift term, so it cannot go negative -- a useful
    cross-check when GK's close-open term is doing something odd."""
    h, l = np.asarray(h, float), np.asarray(l, float)
    v = np.log(h / l) ** 2 / (4.0 * LN2)
    out = np.full(v.shape, np.nan)
    if window <= 0 or window > len(v):
        return out
    csum = np.concatenate([[0.0], np.nancumsum(v)])
    out[window - 1:] = np.sqrt((csum[window:] - csum[:-window]) / window)
    return out


def rogers_satchell_var(o, h, l, c) -> np.ndarray:
    """Drift-robust, non-negative by construction. Unlike GK it stays unbiased
    when the bar has a genuine trend -- which is exactly the regime this label
    is trying to detect."""
    o, h, l, c = (np.asarray(v, dtype=float) for v in (o, h, l, c))
    return (np.log(h / c) * np.log(h / o)) + (np.log(l / c) * np.log(l / o))


def yang_zhang_sigma(o, h, l, c, window: int, k_alpha: float = 1.34) -> np.ndarray:
    """Gap-aware: overnight variance + open-to-close variance + RS drift term.

    This is the one to use on D1 index CFDs, where the maintenance-break gap is
    real risk that GK cannot see.
    """
    o, h, l, c = (np.asarray(v, dtype=float) for v in (o, h, l, c))
    n = len(o)
    out = np.full(n, np.nan)
    if window < 2 or window + 1 > n:
        return out
    log_co = np.log(c / o)
    log_oc_prev = np.concatenate([[np.nan], np.log(o[1:] / c[:-1])])
    rs = rogers_satchell_var(o, h, l, c)
    k = k_alpha / (k_alpha + (window + 1.0) / (window - 1.0))
    for i in range(window, n):
        sl = slice(i - window + 1, i + 1)
        v_o = np.nanvar(log_oc_prev[sl], ddof=1)
        v_c = np.nanvar(log_co[sl], ddof=1)
        v_rs = np.nanmean(rs[sl])
        out[i] = np.sqrt(max(v_o + k * v_c + (1.0 - k) * v_rs, 0.0))
    return out


# --------------------------------------------------------------------- label
def expansion_label(
    high,
    low,
    close,
    atr,
    timestamps,
    trigger_idx,
    k_atr: float = 2.0,
    h_min_hours: float = 48.0,
    h_max_hours: float = 72.0,
    flat_by=None,
):
    """Unsigned 48-72h expansion label for each index in `trigger_idx`.

    Arrays are OLDEST -> NEWEST (numpy convention), which is the REVERSE of the
    MQL5 series convention used in FeatureBuilder.mqh. Getting this backwards
    silently produces a lookahead label, so the orientation is asserted.

    `flat_by(ts) -> datetime64` returns the mandatory liquidation instant for
    the week containing ts (see SwapWindow.mqh for the live counterpart).

    Returns (y, meta); y == -1 marks a DROPPED row, never a class.
    """
    high, low, close, atr = (np.asarray(v, float) for v in (high, low, close, atr))
    ts = np.asarray(timestamps, dtype="datetime64[s]")
    if len(ts) > 1 and ts[0] > ts[-1]:
        raise ValueError("timestamps must run oldest -> newest")

    hmin = np.timedelta64(int(h_min_hours * 3600), "s")
    hmax = np.timedelta64(int(h_max_hours * 3600), "s")

    y, meta = [], []
    for i in trigger_idx:
        entry, a = close[i], atr[i]
        if not np.isfinite(entry) or not np.isfinite(a) or a <= 0.0:
            y.append(-1)
            meta.append({"idx": int(i), "drop": "bad atr/close"})
            continue

        deadline = ts[i] + hmax
        if flat_by is not None:
            deadline = min(deadline, np.datetime64(flat_by(ts[i]), "s"))

        end = int(np.searchsorted(ts, deadline, side="right")) - 1
        if end <= i:
            y.append(-1)
            meta.append({"idx": int(i), "drop": "no path"})
            continue

        # HARD DROP, never a 0: a path cut short by the Friday flat rule is not
        # evidence of chop, it is absence of evidence.
        if ts[end] - ts[i] < hmin:
            y.append(-1)
            meta.append({"idx": int(i), "drop": "clipped below h_min"})
            continue

        path = slice(i + 1, end + 1)
        up = float(np.nanmax(high[path]) - entry)
        dn = float(entry - np.nanmin(low[path]))
        expansion = max(up, dn)
        thr = k_atr * a
        hit = expansion >= thr

        first = -1
        if hit:
            run_up = np.maximum.accumulate(high[path]) - entry
            run_dn = entry - np.minimum.accumulate(low[path])
            reach = np.where(np.maximum(run_up, run_dn) >= thr)[0]
            if len(reach):
                first = int(reach[0]) + 1

        y.append(1 if hit else 0)
        meta.append(
            {
                "idx": int(i),
                "expansion_atr": expansion / a,
                "threshold_atr": k_atr,
                "bars_to_hit": first,
                "path_bars": end - int(i),
                "span_hours": float((ts[end] - ts[i]) / np.timedelta64(1, "h")),
                "up_atr": up / a,
                "dn_atr": dn / a,
            }
        )
    return np.asarray(y), meta


def mean_label_span_bars(meta) -> float:
    """Mean number of bars a label's path covers == the overlap between
    adjacent labels.

    A 72h label on H4 spans 18 bars, so 18 consecutive rows resolve against
    overlapping paths. Feed this into the EXISTING EffectiveSampleSize() /
    PurgeBars() machinery in Labels.mqh -- never report a raw N.
    """
    spans = [m["path_bars"] for m in meta if "path_bars" in m]
    return float(np.mean(spans)) if spans else float("nan")