"""_audit_crabel_stretch_20261017.py - Independent audit of the Opening Volatility Stretch claims of 2026-10-17.

Claims (Antigravity & Gemini 4, CANDIDATES.md 2026-10-17):
  DAX : alpha_factory/hypothesis_alpha_miner.py on cfd_m1/de30eur_m1.parquet, open "07:00 UTC":
        N 2,496, gross +0.3233R, net +0.2933R, t +5.86, 12/12 years, +806.8R.
  NQ  : _test_m1_crabel_validation.py on lse/NQ_F_1m, 09:30-15:55 ET:
        N 2,388, gross +0.1764R, net +0.1514R, t +3.57, 11/11 years, G8 pass 56.65%.

Method: replicate the claimed loops exactly (same rules, per-day numpy instead of iterrows), confirm the
published numbers, then remove one artefact at a time. R unit = the planned stretch (the size the order is
placed with). Report-only, 0 ledger. Tested by: Claude (2026-10-17).
EMPIRICAL RESEARCH CANDIDATE (NOT LAW / NOT DEPLOYED).
"""
import os
import sys

import numpy as np
import pandas as pd

ROOT = r"<repo>"
INBOX = os.path.join(ROOT, "quantlab", "data", "inbox")
LAB = os.path.join(ROOT, "prop_eur2k_lab")
OUT = os.path.join(LAB, "_audit_crabel_stretch_20261017.out")
# [withheld: refers to the lab's private live strategy]
# [withheld: refers to the lab's private live strategy]

_lines = []


def pr(*a):
    s = " ".join(str(x) for x in a)
    print(s, flush=True)
    _lines.append(s)


def tstat(x):
    x = np.asarray(x, float)
    return float(x.mean() / x.std(ddof=1) * np.sqrt(len(x))) if len(x) > 2 else float("nan")


# ----------------------------------------------------------------------------------------------------------
# loaders
# ----------------------------------------------------------------------------------------------------------
def _mid_frame(df):
    m = pd.DataFrame({k: (df[f"bid_{k}"].values + df[f"ask_{k}"].values) / 2.0 for k in ("open", "high", "low", "close")},
                     index=df.index)
    m["spr"] = (df["ask_close"] - df["bid_close"]).values
    return m


def load_dax_as_claimed():
    """hypothesis_alpha_miner.load_dataset: naive Berlin stamps are tz_localize('UTC')'d (no conversion, no -1 min)."""
    df = pd.read_parquet(os.path.join(INBOX, "cfd_m1", "de30eur_m1.parquet"))
    return _mid_frame(df)          # index = naive stamps, read as if UTC


def load_cfd_true_clock(name, tz, session_tz, shift_min=0):
    """lab_utc / data_loader convention: naive exchange-local END-of-minute stamps -> UTC start of minute."""
    df = pd.read_parquet(os.path.join(INBOX, "cfd_m1", f"{name}_m1.parquet"))
    df.columns = [c.lower() for c in df.columns]
    if "date" in df.columns:
        df = df.set_index("date")
    idx = pd.DatetimeIndex(df.index).tz_localize(tz, ambiguous="NaT", nonexistent="NaT")
    ok = ~idx.isna()
    df = df[ok]
    m = _mid_frame(df)
    m.index = idx[ok].tz_convert("UTC") - pd.Timedelta(minutes=1) + pd.Timedelta(minutes=shift_min)
    m = m[~m.index.duplicated(keep="last")].sort_index()
    m.attrs["session_tz"] = session_tz
    return m


def load_lse_nq():
    df = pd.read_parquet(os.path.join(INBOX, "lse", "NQ_F_1m_2016-06-01_2026-10-09.parquet"))
    m = df[["open", "high", "low", "close"]].copy()
    m.index = pd.DatetimeIndex(df["timestamp"])          # UTC, start-of-minute (09:30 ET = volume peak)
    m["spr"] = np.nan
    return m


def keys(m, tz):
    """day key (session-local calendar date) and minute-of-day in the session clock."""
    if tz is None:
        loc = m.index
    else:
        loc = m.index.tz_convert(tz)
    day = loc.normalize().tz_localize(None).values.astype("datetime64[D]") if tz else loc.normalize().values.astype("datetime64[D]")
    hm = (loc.hour * 60 + loc.minute).values
    return day, hm


def stretch_full_day(m, day_key, lag=1):
    """Claimed definition: 10-day mean of min(H-O, O-L) of each calendar day (all quotes), known with `lag` days."""
    g = pd.DataFrame({"o": m["open"].values, "h": m["high"].values, "l": m["low"].values, "d": day_key}).groupby("d")
    d = pd.DataFrame({"o": g["o"].first(), "h": g["h"].max(), "l": g["l"].min()})
    raw = np.minimum(d["h"] - d["o"], d["o"] - d["l"])
    v = raw.rolling(10).mean().shift(lag)
    return dict(zip(pd.DatetimeIndex(v.index).values.astype("datetime64[D]"), v.values))


def stretch_session(m, day_key, hm, open_hm, close_hm, lag=1):
    """Crabel's own definition: stretch from the SESSION open (not the calendar-day open)."""
    sel = (hm >= open_hm) & (hm <= close_hm)
    g = pd.DataFrame({"o": m["open"].values[sel], "h": m["high"].values[sel], "l": m["low"].values[sel],
                      "d": day_key[sel], "t": hm[sel]})
    first_ok = g.groupby("d")["t"].min() == open_hm
    gg = g.groupby("d")
    d = pd.DataFrame({"o": gg["o"].first(), "h": gg["h"].max(), "l": gg["l"].min()})[first_ok]
    raw = np.minimum(d["h"] - d["o"], d["o"] - d["l"])
    v = raw.rolling(10).mean().shift(lag)
    return dict(zip(pd.DatetimeIndex(v.index).values.astype("datetime64[D]"), v.values))


# ----------------------------------------------------------------------------------------------------------
# the claimed loop, vectorised per day, with switchable realism
# ----------------------------------------------------------------------------------------------------------
def simulate(m, day, hm, stretch, open_hm, cut_hm, close_hm, *, gap_fill=False, same_bar_stop=False, mult=1.0):
    O, H, L, C, S = (m[k].values for k in ("open", "high", "low", "close", "spr"))
    sel = np.flatnonzero((hm >= open_hm) & (hm <= close_hm))
    dd = day[sel]
    starts = np.flatnonzero(np.r_[True, dd[1:] != dd[:-1]])
    ends = np.r_[starts[1:], len(sel)]
    rows_out = []
    for a, b in zip(starts, ends):
        r = sel[a:b]
        d = dd[a]
        s = stretch.get(d, np.nan)
        if not np.isfinite(s) or s <= 0 or hm[r[0]] != open_hm:
            continue
        e = s * mult
        op = O[r[0]]
        lt, st = op + e, op - e
        pos, entry, k_in, exit_px, reason, gap = 0, np.nan, -1, np.nan, "", 0.0
        for j, k in enumerate(r):
            if pos == 0:
                if hm[k] <= cut_hm:
                    hl, hs = H[k] >= lt, L[k] <= st
                    if hl and hs:
                        break
                    if hl:
                        pos, k_in = 1, k
                        entry = max(O[k], lt) if gap_fill else lt
                        gap = max(O[k] - lt, 0.0)
                    elif hs:
                        pos, k_in = -1, k
                        entry = min(O[k], st) if gap_fill else st
                        gap = max(st - O[k], 0.0)
                    # entry bar: the order inside one minute is unknown. Lower bound = the stop is hit if the bar also
                    # reaches it; on the open bar itself the open IS the stop, so only a trade THROUGH it counts.
                    strict = j == 0
                    if pos != 0 and same_bar_stop and ((pos == 1 and (L[k] < op if strict else L[k] <= op)) or
                                                       (pos == -1 and (H[k] > op if strict else H[k] >= op))):
                        exit_px, reason, k_out = op, "sl_same_bar", k
                        break
            else:
                if pos == 1 and L[k] <= op:
                    exit_px, reason, k_out = (min(O[k], op) if gap_fill else op), "sl", k
                    break
                if pos == -1 and H[k] >= op:
                    exit_px, reason, k_out = (max(O[k], op) if gap_fill else op), "sl", k
                    break
        if pos == 0:
            continue
        if not reason:
            k_out = r[-1]
            exit_px, reason = C[k_out], "eod"
        R = (exit_px - entry) * pos / e
        if reason == "sl" and not gap_fill:
            R = -1.0                                              # the claimed code books exactly -1.0
        rows_out.append(dict(day=d, year=int(str(d)[:4]), dir=pos, R=R, e=e, px=entry, reason=reason,
                             t_in=int(hm[k_in]), gap_R=gap / e,
                             cost_ba=(0.5 * S[k_in] + 0.5 * S[k_out]) / e))
    T = pd.DataFrame(rows_out)
    if len(T):
        T["day"] = pd.to_datetime(T["day"])
    return T


def summary(T, label, cost_col=None, cost_r=None):
    if len(T) == 0:
        pr(f"  {label}: 0 trades")
        return {}
    c = T[cost_col] if cost_col else pd.Series(cost_r if cost_r is not None else 0.0, index=T.index)
    net = T["R"] - c
    net2 = T["R"] - 2 * c
    yr = net.groupby(T["year"]).sum()
    cum = net.cumsum()
    dd = float((cum.cummax() - cum).max())
    pr(f"  {label}: n {len(T):5d} | gross {T['R'].mean():+.4f} (t {tstat(T['R']):+5.2f}) | cost {c.mean():.4f}R | "
       f"net {net.mean():+.4f} (t {tstat(net):+5.2f}) | 2x {net2.mean():+.4f} (t {tstat(net2):+5.2f}) | "
       f"sum {net.sum():+7.1f}R | years+ {int((yr > 0).sum())}/{len(yr)} | maxDD {dd:.1f}R | win {100 * (T['R'] > 0).mean():.1f}%")
    return dict(n=len(T), gross=T["R"].mean(), net=net.mean(), t=tstat(net), t2=tstat(net2), years=yr)


def years_line(T, cost_col=None, cost_r=0.0):
    c = T[cost_col] if cost_col else cost_r
    yr = (T["R"] - c).groupby(T["year"]).agg(["count", "sum", "mean"])
    pr("    year  " + " ".join(f"{y:>7d}" for y in yr.index))
    pr("    n     " + " ".join(f"{int(v):>7d}" for v in yr["count"]))
    pr("    sum R " + " ".join(f"{v:>+7.1f}" for v in yr["sum"]))
    pr("    mean  " + " ".join(f"{v:>+7.3f}" for v in yr["mean"]))


def ftmo_mc_as_claimed(r_arr, seed=42, n_paths=10000, n_trades=60, block=5, target=10.0, maxdd=10.0):
    """_stress_battery_crabel_stretch.py G8, unchanged logic."""
    rng = np.random.RandomState(seed)
    blocks = [r_arr[i:i + block] for i in range(0, len(r_arr) - block + 1)]
    p = b = 0
    for _ in range(n_paths):
        x = np.concatenate([blocks[i] for i in rng.randint(0, len(blocks), size=n_trades // block)])
        cum = np.cumsum(x)
        ddv = np.maximum.accumulate(cum) - cum
        pi = np.flatnonzero(cum >= target)
        bi = np.flatnonzero(ddv >= maxdd)
        pt = pi[0] if len(pi) else 9999
        bt = bi[0] if len(bi) else 9999
        if pt < bt and pt < len(x):
            p += 1
        elif bt < pt:
            b += 1
    return 100.0 * p / n_paths, 100.0 * b / n_paths


# ----------------------------------------------------------------------------------------------------------
def audit_dax():
    pr("=" * 110)
    pr("PART A - DAX 40 (de30eur_m1.parquet) 'Opening Volatility Stretch'")
    pr("=" * 110)
    m = load_dax_as_claimed()
    day, hm = keys(m, None)
    st = stretch_full_day(m, day)

    pr("A0. Replication of the claimed run. Grid over open/window/close found the published numbers exactly at:")
    pr("    open = bar stamped '08:00', entries while stamp <= '10:00', exit at the bar stamped '17:30', stretch x1.0,")
    pr("    fill at the trigger, stop not checked on the entry bar, flat 0.030R cost (claim: N 2,496, gross +0.3233R).")
    O_HM, CUT_HM, CL_HM = 480, 600, 1050
    T0 = simulate(m, day, hm, st, O_HM, CUT_HM, CL_HM)
    summary(T0, "replication", cost_r=0.030)
    years_line(T0, cost_r=0.030)
    pr("    claim  2015 +188.0  2016 +191.7  2017 +93.8  2018 +117.4  2019 +7.5  2020 +53.2  2021 +53.7  2022 +27.4  "
       "2023 +8.4  2024 +20.9  2025 +34.5  2026 +10.3")

    pr("")
    pr("A1. What the '08:00' bar is. The file is stamped Europe/Berlin local time at the END of the minute (lab_utc.TZ_MAP,")
    pr("    prop_gpu_miner Phase 0); the script reads it as UTC. The bar stamped 08:00 is 07:59-08:00 Berlin: the last")
    pr("    minute BEFORE FDAX opened, until the Eurex extended hours reached this feed. Flat (H = L) share by year:")
    flat = (m["high"].values == m["low"].values)
    yr = m.index.year.values
    for lab_, msk in (("stamps 07:00-07:59", (hm >= 420) & (hm <= 479)), ("stamp 08:00 (open bar)", hm == 480),
                      ("stamps 08:01-08:59", (hm >= 481) & (hm <= 539))):
        s_ = pd.Series(flat[msk]).groupby(yr[msk]).mean()
        pr(f"    {lab_:22s} " + " ".join(f"{y}:{v * 100:4.0f}%" for y, v in s_.items()))
    o_open = pd.Series(m["open"].values[hm == 480], index=day[hm == 480])
    o_next = pd.Series(m["open"].values[hm == 481], index=day[hm == 481])
    jump = (o_next - o_open).abs().dropna()
    stv = pd.Series(st).reindex(jump.index)
    jr = (jump / stv).dropna()
    pr("    |open of 08:01 bar - open of 08:00 bar| in stretch units, median by year: " +
       " ".join(f"{y}:{v:.2f}" for y, v in jr.groupby(pd.DatetimeIndex(jr.index).year).median().items()))

    pr("")
    pr("A2. Where the claimed money comes from: entries whose bar OPENED beyond the trigger (the price jumped over it;")
    pr("    the claimed code still fills at the trigger) and entries in the first FDAX minutes (stamps 08:01-08:05):")
    T0 = T0.copy()
    T0["gap_entry"] = T0["gap_R"] > 0
    T0["fdax_open"] = (T0["t_in"] >= 481) & (T0["t_in"] <= 485)
    for lab, msk in (("gap entries", T0["gap_entry"]), ("no gap", ~T0["gap_entry"]),
                     ("entry 08:01-08:05 stamp", T0["fdax_open"]), ("entry other minutes", ~T0["fdax_open"])):
        sub = T0[msk]
        pr(f"    {lab:26s} n {len(sub):5d}  gross {sub['R'].mean():+.4f}  sum {sub['R'].sum():+7.1f}R  "
           f"free jump at fill {sub['gap_R'].mean():+.3f}R")
    g = T0.groupby("year").agg(n=("R", "size"), gap_share=("gap_entry", "mean"), gap_R=("gap_R", "sum"), R=("R", "sum"))
    pr("    by year: n / % gap entries / sum of free jump R / sum gross R")
    for y, row in g.iterrows():
        pr(f"      {y}  n {int(row.n):4d}  gap {row.gap_share * 100:5.1f}%  jump {row.gap_R:+7.1f}R  gross {row.R:+7.1f}R")
    cut0, close0 = CUT_HM - O_HM, "17:30"

    pr("")
    pr("A3. Same clock and window as the claim, fills made physical one step at a time (cost: the file's own bid/ask):")
    args = (m, day, hm, st, O_HM, CUT_HM, CL_HM)
    Ta = simulate(*args, gap_fill=True)
    summary(Ta, "gap fill at bar open (entry and stop)", cost_r=0.030)
    Tb = simulate(*args, gap_fill=True, same_bar_stop=True)
    summary(Tb, "+ stop checked on the entry bar     ", cost_r=0.030)
    summary(Tb, "+ file bid/ask spread instead of 0.03", cost_col="cost_ba")
    years_line(Tb, cost_col="cost_ba")

    pr("")
    pr("A4. True clock (Berlin local, -1 min) at the real opens, physical fills, file bid/ask cost:")
    mt = load_cfd_true_clock("de30eur", "Europe/Berlin", "Europe/Berlin")
    dayt, hmt = keys(mt, "Europe/Berlin")
    stt = stretch_full_day(mt, dayt)
    res_a4 = {}
    for lab, o_hm, cut, cl in (("FDAX 08:00 Berlin, entry to 09:30, exit 17:30", 480, 570, 1050),
                               ("Xetra 09:00 Berlin, entry to 10:30, exit 17:30", 540, 630, 1050)):
        T = simulate(mt, dayt, hmt, stt, o_hm, cut, cl, gap_fill=True, same_bar_stop=True)
        res_a4[lab] = summary(T, lab, cost_col="cost_ba")
        years_line(T, cost_col="cost_ba")
        sts = stretch_session(mt, dayt, hmt, o_hm, cl)
        T2 = simulate(mt, dayt, hmt, sts, o_hm, cut, cl, gap_fill=True, same_bar_stop=True)
        summary(T2, "   same, Crabel session stretch     ", cost_col="cost_ba")
    pr("")
    pr("A5. The claimed rule (08:00 stamp) only on years when the pre-08:00 quote is live (2019+), physical fills, bid/ask:")
    summary(Tb[Tb["year"] >= 2019], "2019-2026", cost_col="cost_ba")
    summary(Tb[Tb["year"] >= 2022], "2022-2026", cost_col="cost_ba")
    return T0, Tb


def audit_bund():
    pr("")
    pr("=" * 110)
    pr("PART B - German Bund (de10ybeur), the other 'edge' of the priority sweep (t 4.63), same artefact check")
    pr("=" * 110)
    df = pd.read_parquet(os.path.join(INBOX, "cfd_m1", "de10ybeur_m1.parquet"))
    m = _mid_frame(df)
    day, hm = keys(m, None)
    flat = (m["high"].values == m["low"].values)
    yr = m.index.year.values
    s = pd.Series(flat[(hm >= 420) & (hm <= 479)]).groupby(yr[(hm >= 420) & (hm <= 479)]).mean()
    pr("  flat minutes 07:00-07:59 stamp (Eurex Bund opens 08:00 Berlin): " + " ".join(f"{y}:{v * 100:3.0f}%" for y, v in s.items()))
    sts = stretch_session(m, day, hm, 420, 960)
    T = simulate(m, day, hm, sts, 420, 510, 960, mult=0.85)
    summary(T, "claimed (session stretch x0.85, 07:00-16:00 stamp)", cost_r=0.025)
    T["gap_entry"] = T["gap_R"] > 0
    pr(f"    gap entries {T['gap_entry'].mean() * 100:.1f}% of trades carry {T.loc[T['gap_entry'], 'R'].sum():+.1f}R "
       f"of {T['R'].sum():+.1f}R gross (free jump {T['gap_R'].sum():+.1f}R)")
    Tb = simulate(m, day, hm, sts, 420, 510, 960, mult=0.85, gap_fill=True, same_bar_stop=True)
    summary(Tb, "physical fills, bid/ask cost", cost_col="cost_ba")


def audit_nq():
    pr("")
    pr("=" * 110)
    pr("PART C - Nasdaq-100 'Opening Volatility Stretch' (09:30 ET, entry to 11:00, exit 15:55)")
    pr("=" * 110)
    m = load_lse_nq()
    day, hm = keys(m, "America/New_York")
    # the claim maps the UTC-day stretch onto the ET day; for every RTH day that is the same calendar date
    dU = m.index.normalize().tz_localize(None).values.astype("datetime64[D]")
    st = stretch_full_day(m, dU)
    T0 = simulate(m, day, hm, st, 570, 660, 955)
    pr("C0. Replication, LSE NQ.F (clock checked: 09:30 ET bar carries the opening volume, start-stamped):")
    r0 = summary(T0, "claimed rule, 0.025R", cost_r=0.025)
    years_line(T0, cost_r=0.025)
    T0["cost_p1"] = P1_COST_PTS * T0["px"] / P1_COST_REF / T0["e"]
    Tb = simulate(m, day, hm, st, 570, 660, 955, gap_fill=True, same_bar_stop=True)
    Tb["cost_p1"] = P1_COST_PTS * Tb["px"] / P1_COST_REF / Tb["e"]
    # [withheld: refers to the lab's private live strategy]
    # [withheld: refers to the lab's private live strategy]
    Tg = simulate(m, day, hm, st, 570, 660, 955, gap_fill=True)
    Tg["cost_p1"] = P1_COST_PTS * Tg["px"] / P1_COST_REF / Tg["e"]
    Ts = simulate(m, day, hm, st, 570, 660, 955, same_bar_stop=True)
    Ts["cost_p1"] = P1_COST_PTS * Ts["px"] / P1_COST_REF / Ts["e"]
    summary(Tg, "gap fill only, P1 cost             ", cost_col="cost_p1")
    summary(Ts, "entry-bar stop only, P1 cost       ", cost_col="cost_p1")
    summary(Tb, "gap fill + entry-bar stop, P1 cost ", cost_col="cost_p1")
    years_line(Tb, cost_col="cost_p1")
    pr(f"    trades entered on the 09:30 bar itself: {int((T0['t_in'] == 570).sum())}, their claimed gross "
       f"{T0.loc[T0['t_in'] == 570, 'R'].sum():+.1f}R of {T0['R'].sum():+.1f}R")
    pr(f"    same-bar stops: {int((Tb['reason'] == 'sl_same_bar').sum())} trades; median stretch {Tb['e'].median():.1f} pt; "
       f"cost per trade {Tb['cost_p1'].mean():.4f}R")
    pr("C2. Lag test (stretch known with 1 vs 2 days): ")
    st2 = stretch_full_day(m, dU, lag=2)
    T2 = simulate(m, day, hm, st2, 570, 660, 955, gap_fill=True, same_bar_stop=True)
    T2["cost_p1"] = P1_COST_PTS * T2["px"] / P1_COST_REF / T2["e"]
    summary(T2, "lag 2                              ", cost_col="cost_p1")

    pr("")
    pr("C3. Independent data: US100 CFD (nas100usd, Chicago stamps, true clock), 2005-2026. 2005-05..2016-05 was never")
    pr("    seen by the authors; 2016-06+ overlaps the LSE sample.")
    mc = load_cfd_true_clock("nas100usd", "America/Chicago", "America/New_York")
    dayc, hmc = keys(mc, "America/New_York")
    dUc = mc.index.normalize().tz_localize(None).values.astype("datetime64[D]")
    stc = stretch_full_day(mc, dUc)
    Tc = simulate(mc, dayc, hmc, stc, 570, 660, 955, gap_fill=True, same_bar_stop=True)
    Tc["cost_p1"] = P1_COST_PTS * Tc["px"] / P1_COST_REF / Tc["e"]
    summary(Tc, "CFD 2005-2026 all                  ", cost_col="cost_p1")
    pre = Tc[Tc["day"] < pd.Timestamp("2016-06-01")]
    post = Tc[Tc["day"] >= pd.Timestamp("2016-06-01")]
    summary(pre, "CFD 2005-01..2016-05 (unseen)      ", cost_col="cost_p1")
    summary(post, "CFD 2016-06..2026-08 (overlap)     ", cost_col="cost_p1")
    summary(post[post["day"] >= pd.Timestamp("2012-03-01")], "  overlap, file's bid/ask cost     ", cost_col="cost_ba")
    years_line(Tc, cost_col="cost_p1")
    To = simulate(mc, dayc, hmc, stc, 570, 660, 955)
    To["cost_p1"] = P1_COST_PTS * To["px"] / P1_COST_REF / To["e"]
    pr("    upper bound (the claimed optimistic fills: no stop on the entry bar):")
    summary(To, "  CFD 2005-2026 optimistic fills   ", cost_col="cost_p1")
    summary(To[To["day"] < pd.Timestamp("2016-06-01")], "  CFD 2005..2016-05 optimistic     ", cost_col="cost_p1")
    summary(To[To["day"] >= pd.Timestamp("2016-06-01")], "  CFD 2016-06.. optimistic         ", cost_col="cost_p1")
    j = post.merge(Tb, on="day", suffixes=("_cfd", "_lse"))
    pr(f"    overlap match: {len(j)} common days, same direction {100 * (j['dir_cfd'] == j['dir_lse']).mean():.1f}%, "
       f"R correlation {np.corrcoef(j['R_cfd'], j['R_lse'])[0, 1]:.2f}")

    pr("C4. Bar phase on the CFD (rule 9), shift -3..+3 minutes, physical fills, P1 cost:")
    for k in (-3, -2, -1, 0, 1, 2, 3):
        mk = load_cfd_true_clock("nas100usd", "America/Chicago", "America/New_York", shift_min=k)
        dk, hk = keys(mk, "America/New_York")
        sk = stretch_full_day(mk, mk.index.normalize().tz_localize(None).values.astype("datetime64[D]"))
        Tk = simulate(mk, dk, hk, sk, 570, 660, 955, gap_fill=True, same_bar_stop=True)
        Tk["cost_p1"] = P1_COST_PTS * Tk["px"] / P1_COST_REF / Tk["e"]
        summary(Tk, f"phase {k:+d} min                      ", cost_col="cost_p1")

    pr("")
    # [withheld: refers to the lab's private live strategy]
    P = pd.read_csv(P1_LEDGER)
    P["day"] = pd.to_datetime(P["entry_ts"], utc=True).dt.tz_convert("America/New_York").dt.normalize().dt.tz_localize(None)
    p_daily = P.groupby("day")["net"].sum()
    p_dir = P.groupby("day")["dir"].first()
    Tc["net"] = Tc["R"] - Tc["cost_p1"]
    s_daily = Tc.set_index("day")["net"]
    lo, hi = max(p_daily.index.min(), s_daily.index.min()), min(p_daily.index.max(), s_daily.index.max())
    alld = pd.Index(np.unique(np.r_[p_daily.index.values, s_daily.index.values]))
    alld = alld[(alld >= lo) & (alld <= hi)]
    a = p_daily.reindex(alld, fill_value=0.0)
    b = s_daily.reindex(alld, fill_value=0.0)
    shared = p_daily.index.intersection(s_daily.index)
    # [withheld: refers to the lab's private live strategy]
       # [withheld: refers to the lab's private live strategy]
    sd = Tc.set_index("day")["dir"].reindex(shared)
    pr(f"    on shared days same direction: {100 * (sd.values == p_dir.reindex(shared).values).mean():.1f}%; "
       f"daily-R correlation (union of days) {np.corrcoef(a, b)[0, 1]:+.3f}; on shared days {np.corrcoef(a[shared], b[shared])[0, 1]:+.3f}")

    pr("")
    pr("C6. G8 as coded in _stress_battery_crabel_stretch.py, seed sensitivity and corrected trades:")
    r_claim = (T0["R"] - 0.025).values
    for sd_ in (42, 0, 1, 2, 3):
        pp, pb = ftmo_mc_as_claimed(r_claim, seed=sd_)
        pr(f"    claimed trades, seed {sd_:2d}: pass {pp:5.2f}%  bust {pb:5.2f}%")
    r_corr = (Tc["R"] - Tc["cost_p1"]).values
    pp, pb = ftmo_mc_as_claimed(r_corr, seed=42)
    pr(f"    CFD 2005-2026 physical, P1 cost, seed 42: pass {pp:5.2f}%  bust {pb:5.2f}%")
    pr("    note: 'pass' = +10R before a 10R trailing drawdown within 60 trades at 1% risk (Phase 1 only);")
    # [withheld: refers to the lab's private live strategy]
    return T0, Tb, Tc


def audit_sweep_clocks():
    pr("")
    pr("=" * 110)
    pr("PART D - The rest of the 2026-10-17 sweep: what clock each 'UTC' literal really hit")
    pr("=" * 110)
    pr("  hypothesis_alpha_miner.load_dataset localizes every naive index as UTC and never subtracts the end-of-minute")
    pr("  stamp; session times are UTC literals with no DST handling. Real time of the 'open' bar:")
    rows = [
        ("DAX 'standard' (claim)", "de30eur", "Europe/Berlin", "08:00", "07:59 Berlin = 1 min BEFORE FDAX open (stale quote 2015-18)"),
        ("DAX NR-scaled", "de30eur", "Europe/Berlin", "07:00", "06:59 Berlin = 2 h before Xetra, 1 h before FDAX (57% flat 2015-17)"),
        ("Bund 10Y", "de10ybeur", "Europe/Berlin", "07:00", "06:59 Berlin = 1 h BEFORE Eurex Bund opens (84-96% flat 2015-18)"),
        ("US 10Y note", "usb10yusd", "America/Chicago", "12:30", "12:29 Chicago = 13:29 ET, not the 08:20 CT pit open"),
        ("SPX gamma trap", "spx500usd", "America/Chicago", "13:30", "13:29 Chicago = 14:29 ET, afternoon, not the open"),
        ("Dow NR-scaled", "us30usd", "America/Chicago", "13:30", "13:29 Chicago = 14:29 ET, afternoon, not the open"),
        ("Nasdaq NR-scaled", "LSE NQ.F (true UTC)", "UTC", "13:30", "09:30 ET in summer, 08:30 ET in winter (DST not handled)"),
        ("Gold COMEX", "new_markets/xauusd_m1_est", "EST file", "12:30", "12:30 in the file's own zone, not 12:30 UTC"),
        ("EU50 (global script)", "eu50eur", "Europe/Berlin", "07:00", "06:59 Berlin = before the Euro Stoxx cash open"),
        ("JP225 (global script)", "jp225usd", "America/Chicago", "00:00", "23:59 Chicago = mid-session Tokyo afternoon, not 09:00 JST"),
        ("US2000 (global script)", "us2000usd", "America/New_York", "14:30", "14:29 ET = afternoon, not the 09:30 open"),
    ]
    for r in rows:
        pr(f"  {r[0]:24s} {r[1]:26s} file zone {r[2]:17s} literal {r[3]}  ->  {r[4]}")
    pr("  Consequence: the 'kills' of that sweep (US10Y, SPX gamma, gold, S&P stretch timing) did not test the stated")
    pr("  hypotheses; they are VOID, not evidence. The two 'edges' (DAX, Bund) sit exactly on stale pre-open quotes.")

    m = load_dax_as_claimed()
    day, hm = keys(m, None)
    sel = (hm >= 420) & (hm <= 930)
    g = pd.DataFrame({"o": m["open"].values[sel], "h": m["high"].values[sel], "l": m["low"].values[sel], "d": day[sel]}).groupby("d")
    d = pd.DataFrame({"o": g["o"].first(), "h": g["h"].max(), "l": g["l"].min()})
    raw_s = np.minimum(d["h"] - d["o"], d["o"] - d["l"])
    rng = d["h"] - d["l"]
    sk = raw_s.rolling(10).mean().shift(1)
    adr = rng.rolling(10).mean().shift(1)
    eff = sk * np.clip(rng.shift(1) / adr, 0.65, 1.35)
    stn = dict(zip(pd.DatetimeIndex(eff.index).values.astype("datetime64[D]"), eff.values))
    T = simulate(m, day, hm, stn, 420, 510, 930)
    summary(T, "DAX NR-scaled as claimed (07:00 stamp)", cost_r=0.030)
    pr(f"    gap entries {100 * (T['gap_R'] > 0).mean():.1f}% carry {T.loc[T['gap_R'] > 0, 'R'].sum():+.1f}R of {T['R'].sum():+.1f}R gross")
    Tb = simulate(m, day, hm, stn, 420, 510, 930, gap_fill=True, same_bar_stop=True)
    summary(Tb, "physical fills, file bid/ask         ", cost_col="cost_ba")


if __name__ == "__main__":
    parts = sys.argv[1:] or ["dax", "bund", "nq", "clocks"]
    try:
        if "dax" in parts:
            audit_dax()
        if "bund" in parts:
            audit_bund()
        if "nq" in parts:
            audit_nq()
        if "clocks" in parts:
            audit_sweep_clocks()
    finally:
        with open(OUT if parts == ["dax", "bund", "nq", "clocks"] else OUT.replace(".out", f"_{'_'.join(parts)}.out"), "w",
                  encoding="utf-8") as f:
            f.write("\n".join(_lines) + "\n")