PythagorasGoal/scripts/research/blind/verify_top.py

"""verify_top — adversarial second layer on the OOS leaderboard winners.

The auto causality-guard already kills look-ahead. This asks the harder questions the
2026-06-20 sweep taught us to ask before believing ANY directional BTC/ETH edge:

  1. TREND-IN-DISGUISE? Correlate each candidate's OOS net returns to a canonical
     multi-horizon TSMOM (TP01 archetype) on the SAME blind curves. corr>0.7 => it is
     just trend-beta of an up-trending pair, not new alpha.
  2. FEE-ROBUST? Re-score OOS at 0.20% round-trip (4x the per-side baseline). A real
     edge survives; a turnover-churner dies.
  3. STABILITY? Split the OOS tail into K contiguous blocks; drop each in turn and
     recompute Sharpe. Report the worst (jackknife) — a result resting on one block is
     regime-luck, not an edge.

  uv run python scripts/research/blind/verify_top.py [--top 10]
"""
from __future__ import annotations

import argparse
import importlib.util
import json
import sys
from pathlib import Path

import numpy as np

HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))
import blindlib as bl  # noqa: E402

AGENTS = HERE / "agents"


def _sig(path: Path):
    spec = importlib.util.spec_from_file_location(path.stem, path)
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod.signal


def _trend_baseline(df):
    """Canonical TP01-style multi-horizon TSMOM, long-flat, vol-targeted (the thing a
    new directional edge must beat / be orthogonal to)."""
    c = df["close"].values.astype(float)
    r = bl.simple_returns(c)
    sig = np.zeros(len(c))
    for H in (30, 90, 180):
        m = np.zeros(len(c))
        m[H:] = c[H:] / c[:-H] - 1.0
        sig += np.sign(m)
    direction = np.clip(sig / 3.0, 0, 1)        # long-flat
    return bl.vol_target(direction, df, 0.20, 30, 1.0)


def _net(signal_fn, series):
    """OOS net-return vector (test slice) for a signal on a series."""
    df = bl.load(series, "full")
    cut = bl.split_cut(series)
    tgt = np.nan_to_num(np.asarray(signal_fn(df), float), nan=0.0)
    rep = bl.eval_target(df, tgt, bl.FEE_SIDE,
                         metric_mask=np.r_[np.zeros(cut, bool), np.ones(len(df) - cut, bool)])
    # eval_target returns net over the masked region via _metrics; recompute net here
    c = df["close"].values.astype(float)
    r = bl.simple_returns(c)
    pos = np.zeros(len(tgt)); pos[1:] = np.clip(tgt, -1, 1)[:-1]
    net = pos * r - bl.FEE_SIDE * np.abs(np.diff(pos, prepend=0.0))
    return net[cut:], df["datetime"].values[cut:]


def _sharpe(net):
    net = net[np.isfinite(net)]
    return float(np.mean(net) / np.std(net) * np.sqrt(365.25)) if len(net) > 2 and np.std(net) > 0 else 0.0


def _fee_oos_sharpe(signal_fn, series, fee_side):
    df = bl.load(series, "full"); cut = bl.split_cut(series)
    c = df["close"].values.astype(float); r = bl.simple_returns(c)
    tgt = np.clip(np.nan_to_num(np.asarray(signal_fn(df), float)), -1, 1)
    pos = np.zeros(len(tgt)); pos[1:] = tgt[:-1]
    net = pos * r - fee_side * np.abs(np.diff(pos, prepend=0.0))
    return _sharpe(net[cut:])


def verify(name: str) -> dict:
    sig = _sig(AGENTS / f"{name}.py")
    out = {"name": name}
    corrs, jk_worst, fee_sh = [], [], []
    for s in ("A", "B"):
        net, _ = _net(sig, s)
        bnet, _ = _net(_trend_baseline, s)
        m = min(len(net), len(bnet))
        a, b = net[-m:], bnet[-m:]
        mask = np.isfinite(a) & np.isfinite(b)
        corr = float(np.corrcoef(a[mask], b[mask])[0, 1]) if mask.sum() > 3 else 0.0
        corrs.append(corr)
        # jackknife: drop each of K blocks, Sharpe of the rest
        K = 6
        blocks = np.array_split(np.arange(len(net)), K)
        shs = []
        for j in range(K):
            keep = np.concatenate([blocks[k] for k in range(K) if k != j])
            shs.append(_sharpe(net[keep]))
        jk_worst.append(min(shs))
        fee_sh.append(_fee_oos_sharpe(sig, s, 0.001))   # 0.20% RT
    out["corr_to_trend"] = round(float(np.mean(corrs)), 2)
    out["jackknife_worst_sharpe"] = round(float(min(jk_worst)), 2)
    out["fee020_sharpe_min"] = round(float(min(fee_sh)), 2)
    out["verdict"] = (
        "TREND-IN-DISGUISE" if out["corr_to_trend"] > 0.7 else
        "weak/luck" if out["jackknife_worst_sharpe"] < 0.2 else
        "ORTHOGONAL-CANDIDATE")
    return out


def main():
    ap = argparse.ArgumentParser(); ap.add_argument("--top", type=int, default=10)
    args = ap.parse_args()
    lb = json.loads((HERE / "leaderboard.json").read_text())
    top = [r["name"] for r in lb["valid"][:args.top]]
    # baseline self-correlation sanity
    print(f"\n  Adversarial verify of top {len(top)} (corr vs canonical TSMOM trend baseline):\n")
    print(f"  {'strategy':<26} {'corr_trend':>10} {'jk_worst_Sh':>12} {'fee0.20%_Sh':>12}  verdict")
    print(f"  {'-'*78}")
    rows = []
    for name in top:
        v = verify(name); rows.append(v)
        print(f"  {name[:26]:<26} {v['corr_to_trend']:>10.2f} {v['jackknife_worst_sharpe']:>12.2f} "
              f"{v['fee020_sharpe_min']:>12.2f}  {v['verdict']}")
    (HERE / "verify_top.json").write_text(json.dumps(rows, indent=2))
    print(f"\n  -> {HERE/'verify_top.json'}")


if __name__ == "__main__":
    main()