PythagorasGoal/scripts/research/intraday/intra_score.py

"""intra_score — judge an INTRADAY / short-horizon signal with the HARDENED marginal scorer.

New axis (2026-06-21): everything post-reset was 1d. We have certified 5m/15m/1h. A module
defines a CONTINUOUS-position signal:
    def target(df) -> np.array         # per-bar position in [-1,1], decided <= close[i]
    # (or target(df, asset) if you need the asset name, e.g. for DVOL)
This wraps altlib.study_marginal at the chosen TF: it compounds the intraday returns to a
daily series, scores it vs TP01 with the HARDENED gates (multi-cut persistence, in-sample
edge >=0.5, hedge-vs-alpha), AND reports the absolute robustness + FEE SWEEP (0.00-0.20% RT)
+ turnover. Intraday fights fees: a churner dies at 0.20% RT. earns_slot is the bullseye.

  uv run python scripts/research/intraday/intra_score.py --module <path.py> --tf 1h
  uv run python scripts/research/intraday/intra_score.py --all --tf 1h
"""
from __future__ import annotations

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

HERE = Path(__file__).resolve().parent
sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/alt")
import altlib as al  # noqa: E402

AGENTS = HERE / "agents"


def _target(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.target


def score(path: Path, tf: str) -> dict:
    rec = {"name": path.stem, "tf": tf}
    try:
        target = _target(path)
    except Exception as e:
        return {**rec, "error": f"import: {e}", "earns_slot": False}
    try:
        rep = al.study_marginal(path.stem, target, tf=tf)
    except Exception as e:
        import traceback
        return {**rec, "error": f"score: {e}\n{traceback.format_exc()[-300:]}", "earns_slot": False}
    m = rep["marginal"]
    cell = rep["absolute"]["cells"][0]
    # min per-asset turnover/year + worst-case fee Sharpe (0.20% RT)
    turn = max(cell["per_asset"][a]["turnover"] for a in ("BTC", "ETH"))
    fee020 = min(cell["per_asset"][a]["fee_sweep"].get("0.20%RT", -9) for a in ("BTC", "ETH"))
    rec.update(
        abs_grade=rep["abs_grade"], marginal_verdict=rep["marginal_verdict"],
        earns_slot=rep["earns_slot"],
        corr_full=m.get("corr_full"), corr_hold=m.get("corr_hold"),
        uplift_hold=m.get("blends", {}).get("w25", {}).get("uplift_hold"),
        uplift_full=m.get("blends", {}).get("w25", {}).get("uplift_full"),
        cand_insample_sharpe=m.get("cand_insample_sharpe"),
        has_insample_edge=m.get("has_insample_edge"), is_hedge=m.get("is_hedge"),
        robust_oos=m.get("robust_oos"), multicut_persistent=m.get("multicut_persistent"),
        abs_full_sharpe=cell.get("min_asset_full_sharpe"),
        abs_hold_sharpe=cell.get("min_asset_holdout_sharpe"),
        turnover_per_year=round(turn, 1), fee020_full_sharpe=round(fee020, 3),
        fee_survives=cell.get("fee_survives"),
    )
    return rec


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--module"); ap.add_argument("--tf", default="1h")
    ap.add_argument("--all", action="store_true")
    args = ap.parse_args()
    if args.all:
        rows = []
        for p in sorted(AGENTS.glob("agent_*.py")):
            tf = "15m" if "_15m" in p.stem else args.tf
            rows.append(score(p, tf))
        rows.sort(key=lambda r: (r.get("earns_slot", False), r.get("uplift_hold") or -9), reverse=True)
        print(f"\n  INTRADAY wave ({len(rows)} signals) — hardened marginal judge + fee sweep")
        print(f"  {'name':<26}{'tf':>4} {'verdict':<9}{'absG':>5}{'corrH':>6}{'up_h':>6}"
              f"{'is_sh':>6}{'turn':>6}{'fee.20':>7} slot")
        print("  " + "-" * 92)
        for r in rows:
            if "error" in r:
                print(f"  {r['name'][:26]:<26}{r['tf']:>4} ERROR {r['error'][:40]}"); continue
            print(f"  {r['name'][:26]:<26}{r['tf']:>4} {str(r['marginal_verdict']):<9}"
                  f"{str(r['abs_grade']):>5}{str(r.get('corr_hold')):>6}{str(r.get('uplift_hold')):>6}"
                  f"{str(r.get('cand_insample_sharpe')):>6}{str(r.get('turnover_per_year')):>6}"
                  f"{str(r.get('fee020_full_sharpe')):>7}  {'<<<' if r.get('earns_slot') else ''}")
        slots = [r["name"] for r in rows if r.get("earns_slot")]
        print(f"\n  EARNS SLOT: {slots or 'NONE'}")
        (HERE / "intra_leaderboard.json").write_text(json.dumps(rows, indent=2, default=str))
    else:
        print(json.dumps(score(Path(args.module), args.tf), indent=2, default=str))


if __name__ == "__main__":
    main()