fithia2/_tune_v811_scoring.py

"""Sweep v13 scoring variants on v8.11 base. Foreground-safe version."""
import json
import subprocess
import shutil
import sys
from pathlib import Path

BASE = "configs/experiments/return_max_long_v8.11_composed_gld_tqqq_bmc30_cl65_vol40_rmin1_cap96.json"
SWEEP_DIR = Path("configs/experiments/_tune_v811_sweep")
SWEEP_DIR.mkdir(exist_ok=True)
SNAPSHOT = "pead_dualconv_ftb_fix_v2_probe"


def make_config(name: str, scoring_model: str) -> str:
    with open(BASE) as f:
        cfg = json.load(f)
    cfg["experiment_name"] = name
    cfg["overrides"]["signal"]["scoring_model"] = scoring_model
    path = SWEEP_DIR / f"{name}.json"
    path.write_text(json.dumps(cfg, indent=2))
    return str(path)


def run_backtest(config_path: str) -> dict:
    cmd = ["python3", "-m", "apps.backtester.run", "--manifest", config_path,
           "--start", "2022-01-01", "--split", "all", "--initial-equity", "10000",
           "--snapshot-id", SNAPSHOT]
    cache = Path(f"data/parquet/{SNAPSHOT}/.runtime_cache")
    if cache.exists():
        shutil.rmtree(cache)
    proc = subprocess.run(cmd, capture_output=True, text=True, timeout=200)
    tail = proc.stdout.split("\n")[-12:]
    run_id = return_pct = sqs_total = trades = None
    for line in tail:
        if "Run complete:" in line:
            run_id = line.split("Run complete:")[1].strip()
        elif "Total return:" in line:
            try:
                return_pct = float(line.split(":")[1].strip().rstrip("%"))
            except: pass
        elif "SQS:" in line:
            try:
                sqs_total = float(line.split("SQS:")[1].strip().split(" ")[0])
            except: pass
        elif "Trades:" in line:
            try:
                trades = int(line.split(":")[1].strip())
            except: pass
    mdd = sharpe = pf = None
    if run_id:
        m_path = Path(f"runs/{run_id}/metrics/metrics_summary.json")
        if m_path.exists():
            m = json.loads(m_path.read_text())
            mdd = m["max_drawdown_pct"]
            sharpe = m["sharpe_ratio"]
            pf = m["profit_factor"]
    return {"return": return_pct, "trades": trades, "mdd": mdd, "sharpe": sharpe, "pf": pf, "sqs": sqs_total, "run_id": run_id}


SCORING_VARIANTS = [
    "return_max_long_v13e",
    "return_max_long_v13f",
    "return_max_long_v13e_surp",
    "return_max_long_v13e_pd",
    "return_max_long_v13e_mr",
    "return_max_long_v13e_pdmr",
    "return_max_long_v13e_pd2",
    "return_max_long_v13e_h",
    "return_max_long_v13e_ou",
    "return_max_long_v13e_t3",
    "return_max_long_v13e_t23",
    "return_max_long_v13e_qsc",
    "return_max_long_v13e_btg",
    "return_max_long_v13e_hug",
    "return_max_long_v16",
    "return_max_long_v16e",
]


def main():
    # Optional: limit variants via cli arg
    if len(sys.argv) > 1:
        variants = sys.argv[1].split(",")
    else:
        variants = SCORING_VARIANTS

    baseline = {"return": 6839, "mdd": 9.78, "sharpe": 3.53, "pf": 8.26, "sqs": 84.8, "trades": 329}
    print(f"BASELINE v8.11 (v13e): ret={baseline['return']}% MDD={baseline['mdd']}% Sharpe={baseline['sharpe']} PF={baseline['pf']} SQS={baseline['sqs']} trades={baseline['trades']}", flush=True)
    print("=" * 120, flush=True)
    results = []
    for variant in variants:
        name = f"S_{variant.replace('return_max_long_', '')}"
        path = make_config(name, variant)
        print(f"\n>>> {variant}", flush=True)
        try:
            r = run_backtest(path)
            r["variant"] = variant
            if r["return"] is None:
                print(f"    FAILED: no return", flush=True)
                continue
            d_ret = r["return"] - baseline["return"]
            d_sqs = (r["sqs"] or 0) - baseline["sqs"]
            print(f"    ret={r['return']:.0f}% (Δ{d_ret:+.0f}) MDD={r['mdd']:.2f}% Sharpe={r['sharpe']:.2f} PF={r['pf']:.2f} SQS={r['sqs']:.1f} (Δ{d_sqs:+.1f}) trades={r['trades']}", flush=True)
            results.append(r)
        except Exception as e:
            print(f"    FAILED: {e}", flush=True)

    print("\n" + "=" * 120, flush=True)
    print("SUMMARY (sorted by SQS, then return):", flush=True)
    results.sort(key=lambda x: (x.get("sqs") or 0, x.get("return") or 0), reverse=True)
    print(f"{'variant':35s} {'return':>10s} {'MDD':>8s} {'Sharpe':>8s} {'PF':>6s} {'SQS':>7s} {'trades':>7s}", flush=True)
    print(f"{'BASELINE_v13e':35s} {baseline['return']:>9.0f}% {baseline['mdd']:>7.2f}% {baseline['sharpe']:>8.2f} {baseline['pf']:>6.2f} {baseline['sqs']:>7.1f} {baseline['trades']:>7d}", flush=True)
    for r in results:
        print(f"{r['variant']:35s} {r['return']:>9.0f}% {r['mdd']:>7.2f}% {r['sharpe']:>8.2f} {r['pf']:>6.2f} {r['sqs']:>7.1f} {r['trades']:>7d}", flush=True)


if __name__ == "__main__":
    main()