PythagorasGoal/scripts/research/skyhook/runs/SKH2_PCTL_DD.py

"""SKH2_PCTL_DD — DD-reduction wave, family [PCTL_DD].

GOAL: cut STANDALONE maxDD below 30% (max over BTC & ETH) while keeping minHold>=~0.70
and earns_slot==True, using the CAUSAL expanding/rolling PERCENTILE-RANK regime from
SKH_R_PCTL.py (reuse pctl_entries), tuned together with the winner's exits.

Baseline to beat (V2 winner, Chande regime):
  SkyhookParams(ptn_n=45, sl_atr=2.5, tp_atr=7.0, uscitalong=24, uscitashort=16,
                vola_lo=35, vola_hi=95, vol_lo=0.0)
  minFull +0.83, minHold +0.81, standalone DD BTC34%/ETH31% (THE PROBLEM),
  marginal ADDS, blend w25 uplift_hold +0.58, blend 50/50 full1.59/hold1.04/DD12.5%.

LEVERS FOR DD CUT (all causal, expressed through pctl_entries cfg + the SkyhookParams exits):
  * percentile-rank regime bands (where ATR/volume sit in their own causal history):
      - cap the upper vola band (avoid blow-off-vol entries that cluster losses)
      - add a volume floor (live tape only) OR keep vol open
  * tighter hard stop (sl_atr) caps per-trade loss -> shrinks DD
  * the winner's wider tp_atr=7.0 + asym time exits (24/16) carried over.

A candidate BEATS THE WINNER iff: earns_slot AND max_dd<0.30 AND blend_w25_uplift_hold>=0.55
AND min_hold_sharpe>=0.65.  We report TRUE numbers regardless.
"""
from __future__ import annotations

import sys
sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/skyhook")

import importlib.util
import numpy as np
import pandas as pd

import skyhooklib as sk
import altlib as al
from src.backtest.harness import backtest_signals
from src.strategies import skyhook as S
from src.strategies.skyhook import SkyhookParams

# import the structural pctl builder (pctl_entries, pctl_rank, _split) from the sweep script
spec = importlib.util.spec_from_file_location(
    "skr", "/opt/docker/PythagorasGoal/scripts/research/skyhook/runs/SKH_R_PCTL.py")
skr = importlib.util.module_from_spec(spec)
spec.loader.exec_module(skr)   # __main__ guard prevents the sweep from running

HOLDOUT = sk.HOLDOUT
FEE = sk.FEE_RT


# ---------------------------------------------------------------------------
# Build a SkyhookParams holding the WINNER's exits; only regime comes from pctl cfg.
# pctl_entries reads: ptn_n, sl_atr, tp_atr, uscitalong, uscitashort, exit_mode, ltf_atr_win,
#                     max_per_day, long_only  (the regime bands come from the cfg kwargs).
# ---------------------------------------------------------------------------
def winner_exit_params(**kw):
    base = dict(ptn_n=45, sl_atr=2.5, tp_atr=7.0, uscitalong=24, uscitashort=16)
    base.update(kw)
    return SkyhookParams(**base)


# ---------------------------------------------------------------------------
# Eval a (cfg, params) pair on both assets: FULL + HOLD via the honest engine.
# ---------------------------------------------------------------------------
def eval_pair(cfg, p):
    out = {}
    for a in ("BTC", "ETH"):
        ltf, htf = sk.frames(a)
        ent = skr.pctl_entries(ltf, htf, p, **cfg)
        m = backtest_signals(ltf, ent, fee_rt=FEE, leverage=1.0, asset=a, tf="230m")
        eq = m.equity
        idx = pd.DatetimeIndex(pd.to_datetime(m.eq_index, utc=True))
        hmask = np.asarray(idx >= HOLDOUT)
        full = dict(sharpe=round(m.sharpe, 3), ret=round(m.net_return, 4), maxdd=round(m.max_dd, 4),
                    n_trades=int(m.n_trades), win_rate=round(m.win_rate, 1))
        hold = skr._split(eq, idx, hmask)
        out[a] = dict(full=full, hold=hold,
                      yearly={int(y): round(v, 4) for y, v in m.yearly.items()})
    return out


def summarize(res):
    mf = min(res[a]["full"]["sharpe"] for a in res)
    mh = min(res[a]["hold"]["sharpe"] for a in res)
    mt = min(res[a]["full"]["n_trades"] for a in res)
    mdd = max(res[a]["full"]["maxdd"] for a in res)
    return dict(minFull=mf, minHold=mh, minTr=mt, maxDD=mdd, res=res)


def line(tag, v):
    r = v["res"]
    print(f"  [{tag:30s}] minFull={v['minFull']:+.2f} minHold={v['minHold']:+.2f} "
          f"minTr={v['minTr']:3d} maxDD={v['maxDD']*100:4.0f}% | "
          f"BTC F{r['BTC']['full']['sharpe']:+.2f}/H{r['BTC']['hold']['sharpe']:+.2f}/DD{r['BTC']['full']['maxdd']*100:.0f}% "
          f"ETH F{r['ETH']['full']['sharpe']:+.2f}/H{r['ETH']['hold']['sharpe']:+.2f}/DD{r['ETH']['full']['maxdd']*100:.0f}%")


# ---------------------------------------------------------------------------
# Causality (truncated-prefix) on the structural pctl entries.
# ---------------------------------------------------------------------------
def check_causality(cfg, p, asset="BTC", tail=200):
    return skr.check_causality(cfg, p, asset, tail=tail)


# ---------------------------------------------------------------------------
# Marginal vs TP01 on a (cfg, params) pair (50/50 daily, same convention as skyhooklib).
# ---------------------------------------------------------------------------
def marginal_struct(cfg, p):
    def daily(a):
        ltf, htf = sk.frames(a)
        ent = skr.pctl_entries(ltf, htf, p, **cfg)
        m = backtest_signals(ltf, ent, fee_rt=FEE, leverage=1.0, asset=a, tf="230m")
        s = pd.Series(m.equity, index=pd.DatetimeIndex(pd.to_datetime(m.eq_index, utc=True)))
        return s.resample("1D").last().ffill().pct_change().dropna()
    sb, se = daily("BTC"), daily("ETH")
    J = pd.concat({"BTC": sb, "ETH": se}, axis=1, join="inner").fillna(0.0)
    cand = 0.5 * J["BTC"] + 0.5 * J["ETH"]
    return al.marginal_vs_tp01(cand)


def fee_sweep(cfg, p):
    ok = True
    rows = {}
    for a in ("BTC", "ETH"):
        ltf, htf = sk.frames(a)
        ent = skr.pctl_entries(ltf, htf, p, **cfg)
        row = []
        for f in (0.0, 0.001, 0.002, 0.003):
            m = backtest_signals(ltf, ent, fee_rt=f, leverage=1.0, asset=a, tf="230m")
            row.append((f, round(m.sharpe, 3)))
        rows[a] = row
        ok = ok and (dict(row)[0.003] > 0)
    return ok, rows


if __name__ == "__main__":
    print("=== SKH2_PCTL_DD : percentile-rank regime tuned for DD<30 ===\n")

    # -----------------------------------------------------------------------
    # STAGE 1 — coarse sweep: regime bands (pctl space) x stop tightness.
    # Winner exits (tp7/24/16) carried; we vary sl_atr and the regime cfg.
    # Intuition for DD cut:
    #   - cap vola_hi (drop blow-off-vol entries) ; modest vol floor (live tape)
    #   - tighter sl_atr (2.0/1.8) caps per-trade loss.
    # -----------------------------------------------------------------------
    print("--- STAGE 1: regime band x stop sweep (exits tp7/24/16) ---")
    band_cfgs = {
        # name: pctl regime cfg (expanding unless _r)
        "volaHi95_vol0":   dict(vola_win=None, vol_win=None, vola_lo=0.35, vola_hi=0.95, vol_lo=0.0,  vol_hi=1.0),
        "volaHi90_vol0":   dict(vola_win=None, vol_win=None, vola_lo=0.35, vola_hi=0.90, vol_lo=0.0,  vol_hi=1.0),
        "volaMid_vol0":    dict(vola_win=None, vol_win=None, vola_lo=0.25, vola_hi=0.85, vol_lo=0.0,  vol_hi=1.0),
        "volaMid_volFlr":  dict(vola_win=None, vol_win=None, vola_lo=0.25, vola_hi=0.85, vol_lo=0.30, vol_hi=1.0),
        "volaHi90_volFlr": dict(vola_win=None, vol_win=None, vola_lo=0.35, vola_hi=0.90, vol_lo=0.30, vol_hi=1.0),
        "volaCap80_volFlr":dict(vola_win=None, vol_win=None, vola_lo=0.30, vola_hi=0.80, vol_lo=0.30, vol_hi=1.0),
    }
    sls = [2.5, 2.0, 1.8]
    stage1 = {}
    for bname, cfg in band_cfgs.items():
        for sl in sls:
            p = winner_exit_params(sl_atr=sl)
            tag = f"{bname}|sl{sl}"
            v = summarize(eval_pair(cfg, p))
            stage1[tag] = (cfg, p, v)
            line(tag, v)

    # Pick DD<30 candidates with the best minHold (need minHold>=~0.7).
    sub30 = {t: tup for t, tup in stage1.items() if tup[2]["maxDD"] < 0.30}
    print(f"\n--- STAGE 1: configs with maxDD<30%: {len(sub30)} ---")
    for t, (_, _, v) in sorted(sub30.items(), key=lambda kv: -kv[1][2]["minHold"]):
        line(t, v)

    # -----------------------------------------------------------------------
    # STAGE 2 — refine: take best DD<30 (and near-30 with high hold) candidates,
    # fine-tune bands/stop to push minHold up while keeping DD<30.
    # -----------------------------------------------------------------------
    print("\n--- STAGE 2: refinement around best DD<30 / high-hold cells ---")
    refine = {
        # tighter blow-off cap + small vol floor, sl 1.8-2.0
        "R1": (dict(vola_win=None, vol_win=None, vola_lo=0.30, vola_hi=0.85, vol_lo=0.20, vol_hi=1.0), winner_exit_params(sl_atr=2.0)),
        "R2": (dict(vola_win=None, vol_win=None, vola_lo=0.30, vola_hi=0.85, vol_lo=0.20, vol_hi=1.0), winner_exit_params(sl_atr=1.8)),
        "R3": (dict(vola_win=None, vol_win=None, vola_lo=0.30, vola_hi=0.88, vol_lo=0.30, vol_hi=1.0), winner_exit_params(sl_atr=2.0)),
        "R4": (dict(vola_win=None, vol_win=None, vola_lo=0.35, vola_hi=0.85, vol_lo=0.30, vol_hi=1.0), winner_exit_params(sl_atr=2.0)),
        # rolling-window regime (recent), which reacts faster to regime shift -> may cut DD
        "R5": (dict(vola_win=120, vol_win=120, vola_lo=0.30, vola_hi=0.85, vol_lo=0.20, vol_hi=1.0), winner_exit_params(sl_atr=2.0)),
        "R6": (dict(vola_win=120, vol_win=120, vola_lo=0.30, vola_hi=0.85, vol_lo=0.30, vol_hi=1.0), winner_exit_params(sl_atr=1.8)),
        # tighter tp to bank faster (lower DD) with tight sl
        "R7": (dict(vola_win=None, vol_win=None, vola_lo=0.30, vola_hi=0.85, vol_lo=0.20, vol_hi=1.0), winner_exit_params(sl_atr=2.0, tp_atr=6.0)),
        "R8": (dict(vola_win=None, vol_win=None, vola_lo=0.30, vola_hi=0.85, vol_lo=0.20, vol_hi=1.0), winner_exit_params(sl_atr=1.6)),
    }
    stage2 = {}
    for t, (cfg, p) in refine.items():
        v = summarize(eval_pair(cfg, p))
        stage2[t] = (cfg, p, v)
        line(t, v)

    # -----------------------------------------------------------------------
    # PICK BEST: among ALL cells, prefer maxDD<0.30 AND minHold>=0.65; rank by
    # (DD<30) then minHold then -DD. Fall back to best minHold if none sub-30.
    # -----------------------------------------------------------------------
    allcells = {**stage1, **stage2}
    def score(tup):
        v = tup[2]
        dd_ok = v["maxDD"] < 0.30
        hold_ok = v["minHold"] >= 0.65
        full_ok = v["minFull"] >= 0.5
        tr_ok = v["minTr"] >= 20
        # primary: meets all gates; secondary: minHold; tertiary: lower DD
        return (dd_ok and hold_ok and full_ok and tr_ok, dd_ok, v["minHold"], -v["maxDD"])
    best_tag = max(allcells, key=lambda t: score(allcells[t]))
    best_cfg, best_p, best_v = allcells[best_tag]
    print(f"\n*** SELECTED = {best_tag} ***")
    line(best_tag, best_v)

    # -----------------------------------------------------------------------
    # FULL VERIFICATION on selected: causality + fee sweep + marginal.
    # -----------------------------------------------------------------------
    print("\n--- causality (truncated-prefix) ---")
    cB = check_causality(best_cfg, best_p, "BTC")
    cE = check_causality(best_cfg, best_p, "ETH")
    causality_ok = bool(cB["ok"] and cE["ok"])
    print(f"  BTC={cB}  ETH={cE}  -> causality_ok={causality_ok}")

    print("\n--- fee sweep (FULL sharpe) ---")
    fee_ok, frows = fee_sweep(best_cfg, best_p)
    for a, row in frows.items():
        print(f"  {a}: " + " ".join(f"{f*100:.2f}%={s:+.2f}" for f, s in row))
    print(f"  fee_survives 0.30%RT (both): {fee_ok}")

    print("\n--- marginal vs TP01 (selected) ---")
    marg = marginal_struct(best_cfg, best_p)
    corr_full = marg.get("corr_full")
    verdict = marg.get("marginal_verdict")
    has_edge = marg.get("has_insample_edge")
    is_hedge = marg.get("is_hedge")
    robust_oos = marg.get("robust_oos")
    multicut = marg.get("multicut_persistent")
    clean_yr = marg.get("clean_year_uplift")
    w25 = marg.get("blends", {}).get("w25", {})
    w50 = marg.get("blends", {}).get("w50", {})
    up_h = w25.get("uplift_hold")
    print(f"  corr_full={corr_full} corr_hold={marg.get('corr_hold')}")
    print(f"  marginal_verdict={verdict} robust_oos={robust_oos} multicut_persistent={multicut}")
    print(f"  has_insample_edge={has_edge} is_hedge={is_hedge} cand_insample_sharpe={marg.get('cand_insample_sharpe')}")
    print(f"  clean_year_uplift={clean_yr} jackknife_min_uplift={marg.get('jackknife_min_uplift')}")
    print(f"  blend w25: uplift_hold={up_h} uplift_full={w25.get('uplift_full')}")
    print(f"  blend w50: full={w50.get('full')} hold={w50.get('hold')} uplift_hold={w50.get('uplift_hold')} dd={w50.get('dd')}")

    # grade (mirror sk verdict thresholds): PASS if minTr>=20 & minFull>=0.5 & minHold>=0.2 & feeOK
    mf, mh, mt, mdd = best_v["minFull"], best_v["minHold"], best_v["minTr"], best_v["maxDD"]
    if mt >= 20 and mf >= 0.5 and mh >= 0.2 and fee_ok:
        grade = "PASS"
    elif mt >= 20 and mf >= 0.3 and mh >= 0.0:
        grade = "WEAK"
    else:
        grade = "FAIL"
    earns_slot = (grade != "FAIL") and (verdict == "ADDS") and bool(robust_oos) and (not bool(is_hedge))
    beats_winner = bool(earns_slot and mdd < 0.30 and (up_h is not None and up_h >= 0.55) and mh >= 0.65)

    print("\n" + "=" * 70)
    print("FINAL BLOCK — SKH2_PCTL_DD")
    print("=" * 70)
    print(f"best_cfg(regime) = {best_cfg}")
    print(f"best_params      = ptn_n={best_p.ptn_n} sl_atr={best_p.sl_atr} tp_atr={best_p.tp_atr} "
          f"uscitalong={best_p.uscitalong} uscitashort={best_p.uscitashort} exit_mode={best_p.exit_mode}")
    print(f"grade={grade}")
    print(f"minFull={mf:+.3f}  minHold={mh:+.3f}  max_dd={mdd:.4f} ({mdd*100:.0f}%)  n_trades_min={mt}")
    print(f"fee@0.30%RT survives={fee_ok}  causality_ok={causality_ok}")
    print(f"marginal: verdict={verdict} corr_full={corr_full} has_insample_edge={has_edge} "
          f"is_hedge={is_hedge} robust_oos={robust_oos} multicut_persistent={multicut} clean_year_uplift={clean_yr}")
    print(f"blend w25 uplift_hold={up_h}  blend w50 full={w50.get('full')}/hold={w50.get('hold')}/dd={w50.get('dd')}")
    print(f"earns_slot={earns_slot}")
    print(f"beats_winner={beats_winner}")
    print("=" * 70)

    # machine-readable echo for the agent
    import json
    print("RESULT_JSON=" + json.dumps({
        "best_cfg": best_cfg,
        "best_params": {"ptn_n": best_p.ptn_n, "sl_atr": best_p.sl_atr, "tp_atr": best_p.tp_atr,
                        "uscitalong": best_p.uscitalong, "uscitashort": best_p.uscitashort,
                        "exit_mode": best_p.exit_mode,
                        "vola_lo": best_cfg["vola_lo"], "vola_hi": best_cfg["vola_hi"],
                        "vol_lo": best_cfg["vol_lo"], "vol_hi": best_cfg["vol_hi"],
                        "vola_win": best_cfg["vola_win"], "vol_win": best_cfg["vol_win"]},
        "grade": grade, "minFull": mf, "minHold": mh, "max_dd": mdd, "n_trades_min": mt,
        "fee_ok": fee_ok, "causality_ok": causality_ok,
        "marginal_verdict": verdict, "corr_full": corr_full, "has_insample_edge": has_edge,
        "is_hedge": is_hedge, "robust_oos": robust_oos, "multicut_persistent": multicut,
        "clean_year_uplift": clean_yr, "blend_w25_uplift_hold": up_h,
        "w50_full": w50.get("full"), "w50_hold": w50.get("hold"), "w50_dd": w50.get("dd"),
        "earns_slot": earns_slot, "beats_winner": beats_winner,
    }, default=str))