PythagorasGoal/scripts/research/xsec/verify_survivors.py

"""verify_survivors — adversarial 'Verify' phase for the xsec sweep (2026-06-20).

The Find phase flagged 42/257 cross-sectional configs as earns_slot=True on the certified
Hyperliquid panel. ALL the slot-earners share two tells: (a) strongly NEGATIVE corr to TP01
(-0.2..-0.4), (b) PnL concentrated in 2025. Hypothesis under test (the only thing that matters
before promoting any of them to a sleeve):

    "These are not N independent edges. They are ONE regime bet — short the high-beta alt junk
     during the 2024-26 alt-bear — wearing many masks (low-vol, low-beta, low-corr, reversal,
     trend-gated-mom). The drop-one-month jackknife is robust only WITHIN that single regime."

Three skeptics, deterministic (no agents):
  S1 (distinctness/redundancy): mutual correlation matrix of the strongest survivor per family.
     If they're all mutually >0.6 correlated -> one bet, not many.
  S2 (short-beta tell): correlation of each survivor to two reference factors built on the SAME
     panel: SHORTBETA = book ranking by -roll_beta; SHORTMKT = -equal-weight alt-market return.
     A genuinely market-neutral factor should NOT load heavily on "short the market".
  S3 (single-regime): per-calendar-year Sharpe. If the edge is ~entirely 2025, the 2.5y panel
     has ONE up-for-the-factor regime and the hold-out (2025-26) cannot prove robustness.

Run: uv run python scripts/research/xsec/verify_survivors.py
"""
import sys
from pathlib import Path

import numpy as np
import pandas as pd

sys.path.insert(0, str(Path(__file__).resolve().parent))
import xslib as xs  # noqa: E402
import altlib as al  # noqa: E402  (via xslib sys.path)


def roll_corr_to_market(ret, win):
    """Rolling corr of each asset's return to the equal-weight market (causal)."""
    mkt = pd.Series(xs.market_ret(ret))
    out = np.full_like(ret, np.nan)
    for a in range(ret.shape[1]):
        out[:, a] = pd.Series(ret[:, a]).rolling(win, min_periods=max(5, win // 2)).corr(mkt).values
    return out


def book(universe, score_fn, H=10, k=5, long_short=True):
    p = xs.load_panel(universe)
    return xs.xs_backtest(p, score_fn(p), H=H, k=k, long_short=long_short)


# ── strongest representative survivor per family (from the Find-phase output) ──
SURV = {
    "XV02_lowidiovol":  lambda: book("majors", lambda P: -xs.roll_std(xs.residual_return(P.ret, 60), 30)),
    "XV01_lowvol":      lambda: book("majors", lambda P: -xs.roll_std(P.ret, 30)),
    "XV03_lowbeta":     lambda: book("all",    lambda P: -xs.roll_beta(P.ret, 60)),
    "XS06b_lowcorr":    lambda: book("all",    lambda P: -roll_corr_to_market(P.ret, 60)),
    "XU02_lowvol_maj":  lambda: book("majors", lambda P: -xs.roll_std(P.ret, 30), k=5),
    "XM09_trendgmom":   lambda: book("all",    lambda P: _trend_gated_mom(P, 60)),
    "XL02_voltrendmom": lambda: book("majors", lambda P: xs.xs_zscore(xs.past_return(P.close, 60)) + xs.xs_zscore(xs.volume_z(P.vol, 30))),
    "XR02_revgated":    lambda: book("majors", lambda P: _vol_gated_rev(P, 3), H=3),
}


def _trend_gated_mom(P, L):
    """XS momentum, but zeroed on days the equal-weight market trailing-sum is non-positive."""
    s = xs.past_return(P.close, L)
    mkt = xs.market_ret(P.ret)
    up = pd.Series(mkt).rolling(L, min_periods=L // 2).sum().values > 0
    out = s.copy()
    out[~up, :] = np.nan          # flat (no ranking) when market not trending up
    return out


def _vol_gated_rev(P, L):
    """Short-term reversal, active only when market realized vol is in its high regime."""
    rev = -xs.past_return(P.close, L)
    mvol = pd.Series(xs.market_ret(P.ret)).rolling(20, min_periods=10).std()
    thr = mvol.expanding(min_periods=60).quantile(0.70).values
    hi = (mvol.values > thr)
    out = rev.copy()
    out[~hi, :] = np.nan
    return out


# reference factors (the suspected single underlying bet)
REF = {
    "SHORTBETA":  lambda: book("all", lambda P: -xs.roll_beta(P.ret, 60)),   # explicit short-high-beta
    "SHORTMKT":   None,                                                       # -equal-weight alt market
}


def main():
    print("=" * 96)
    print("  ADVERSARIAL VERIFY — are the xsec survivors one regime bet (short alt-beta) or N edges?")
    print("=" * 96)

    series = {n: al._to_daily(f()) for n, f in SURV.items()}

    # SHORTMKT reference = negative equal-weight alt-market daily return (vol-targeted like a book)
    p_all = xs.load_panel("all")
    mkt = pd.Series(-xs.market_ret(p_all.ret), index=p_all.index)
    series["SHORTBETA"] = al._to_daily(book("all", lambda P: -xs.roll_beta(P.ret, 60)))
    series["SHORTMKT"] = al._to_daily(mkt)

    df = pd.DataFrame(series).dropna(how="all")

    names = list(SURV.keys())
    # ── S1: mutual correlation matrix ─────────────────────────────────────────
    print("\n[S1] Mutual correlation matrix of survivors (>0.6 = same bet):")
    C = df[names].corr()
    hdr = "                   " + " ".join(f"{n[:8]:>8s}" for n in names)
    print(hdr)
    for n in names:
        row = " ".join(f"{C.loc[n, m]:>8.2f}" for m in names)
        print(f"  {n:<16s} {row}")
    iu = [(a, b) for i, a in enumerate(names) for b in names[i + 1:]]
    pair_corrs = [C.loc[a, b] for a, b in iu]
    print(f"  --> mean off-diagonal corr = {np.mean(pair_corrs):+.2f}  "
          f"(share |r|>0.6: {np.mean([abs(x) > 0.6 for x in pair_corrs]) * 100:.0f}%)")

    # ── S2: load on short-beta / short-market ─────────────────────────────────
    print("\n[S2] Correlation of each survivor to the suspected single bet:")
    print(f"  {'survivor':<18s} {'corr_SHORTBETA':>15s} {'corr_SHORTMKT':>15s}")
    for n in names:
        cb = df[n].corr(df["SHORTBETA"])
        cm = df[n].corr(df["SHORTMKT"])
        print(f"  {n:<18s} {cb:>15.2f} {cm:>15.2f}")

    # ── S3: per-year Sharpe (single-regime test) ──────────────────────────────
    print("\n[S3] Per-calendar-year Sharpe (is the edge ~entirely 2025?):")
    print(f"  {'survivor':<18s} {'2024':>8s} {'2025':>8s} {'2026':>8s}")
    for n in names:
        s = df[n].dropna()
        cells = []
        for y in (2024, 2025, 2026):
            sy = s[s.index.year == y]
            cells.append(f"{al._sh(sy):>8.2f}" if len(sy) > 20 else f"{'--':>8s}")
        print(f"  {n:<18s} " + " ".join(cells))

    print("\n" + "=" * 96)
    print("  VERDICT logic: high mutual corr + high SHORTBETA/SHORTMKT load + 2025-only Sharpe")
    print("  => one short-alt-beta regime bet on a single-regime 2.5y panel. LEAD/forward-monitor,")
    print("     NOT a sleeve (cannot prove it survives an alt-bull regime flip).")
    print("=" * 96)


if __name__ == "__main__":
    main()