PythagorasGoal/scripts/research/xsec/runs/XS08b.py

"""XS08b — Lead-lag vs BTC.

IDEA: Score = past_return(alt, L=10) of alts CONDITIONAL on BTC having risen over the same
window. The hypothesis: alts that lagged BTC during a BTC up-move will catch up.

Score at bar i:
  btc_ret_L  = BTC.close[i] / BTC.close[i-L] - 1       (BTC rose L days ago to now)
  alt_ret_L  = alt.close[i] / alt.close[i-L] - 1        (how much alt has moved)
  If btc_ret_L > 0:
      score = alt_ret_L (lag = low score -> buy the laggards -> REVERSE ranking needed)
      Actually: we want alts that HAVEN'T moved yet, i.e. low alt_ret when BTC is up.
      So score = -alt_ret_L (lower alt return during BTC up = more upside potential).
  If btc_ret_L <= 0:
      score = NaN (flat; no lead-lag expected when BTC is down).

Alternative formulation (XS08b-v2): score = btc_ret - alt_ret (gap; higher = more lag = more catch-up).

Grid (<=5 calls):
  1. L=10, majors,  H=10, k=5, long_short=True  — baseline
  2. L=10, majors,  H=5,  k=5, long_short=True  — faster rebalance
  3. L=10, "all",   H=10, k=5, long_short=True  — wider universe
  4. L=10, majors,  H=10, k=5, long_short=False — long-only variant
  5. L=20, majors,  H=10, k=5, long_short=True  — longer lookback
"""
import sys
sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/xsec")
import xslib as xs
import numpy as np

# ---------------------------------------------------------------------------
# Score factory
# ---------------------------------------------------------------------------
def make_score(L=10):
    """Score: BTC-alt gap during BTC up-moves. Causal."""
    def score_fn(P: xs.Panel) -> np.ndarray:
        syms = P.syms
        n, A = P.close.shape

        # BTC column index (BTC should be in the majors panel)
        if "BTC" not in syms:
            raise ValueError("BTC not in panel — use 'majors' or a universe containing BTC")
        btc_idx = syms.index("BTC")

        # past return over L days (causal)
        pr = xs.past_return(P.close, L)  # (n, A)

        btc_pr = pr[:, btc_idx]  # (n,) BTC L-day return

        # score = BTC_return - alt_return  (gap; higher gap = alt lagged more = more catch-up)
        # Only when BTC is up (btc_pr > 0); else NaN (flat)
        score = np.full((n, A), np.nan)
        btc_up = btc_pr > 0  # (n,) boolean mask
        gap = btc_pr[:, None] - pr  # (n, A): positive when alt lagged BTC
        score[btc_up] = gap[btc_up]

        return score
    return score_fn


# ---------------------------------------------------------------------------
# Grid
# ---------------------------------------------------------------------------
results = []

print("=" * 60)
print("XS08b — Lead-lag vs BTC")
print("=" * 60)

# 1. Baseline: L=10, majors, H=10, k=5, long_short
print("\n[1/5] L=10, majors, H=10, k=5, long_short=True")
r1 = xs.study_xs("XS08b-base", make_score(L=10),
                  universe="majors", H=10, k=5, long_short=True)
print(xs.fmt(r1))
print("JSON:", xs.as_json(r1))
results.append(r1)

# 2. Faster rebalance: H=5
print("\n[2/5] L=10, majors, H=5, k=5, long_short=True")
r2 = xs.study_xs("XS08b-H5", make_score(L=10),
                  universe="majors", H=5, k=5, long_short=True)
print(xs.fmt(r2))
print("JSON:", xs.as_json(r2))
results.append(r2)

# 3. Wider universe: all
print("\n[3/5] L=10, all, H=10, k=5, long_short=True")
r3 = xs.study_xs("XS08b-all", make_score(L=10),
                  universe="all", H=10, k=5, long_short=True)
print(xs.fmt(r3))
print("JSON:", xs.as_json(r3))
results.append(r3)

# 4. Long-only: majors, H=10
print("\n[4/5] L=10, majors, H=10, k=5, long_short=False")
r4 = xs.study_xs("XS08b-LO", make_score(L=10),
                  universe="majors", H=10, k=5, long_short=False)
print(xs.fmt(r4))
print("JSON:", xs.as_json(r4))
results.append(r4)

# 5. Longer lookback: L=20
print("\n[5/5] L=20, majors, H=10, k=5, long_short=True")
r5 = xs.study_xs("XS08b-L20", make_score(L=20),
                  universe="majors", H=10, k=5, long_short=True)
print(xs.fmt(r5))
print("JSON:", xs.as_json(r5))
results.append(r5)

# ---------------------------------------------------------------------------
# Pick best by: earns_slot first, then hold-out Sharpe, then corr_xs01 < 0.6
# ---------------------------------------------------------------------------
def score_result(r):
    earns = r.get("earns_slot", False)
    ho = (r.get("holdout") or {}).get("sharpe", -999)
    full = (r.get("full") or {}).get("sharpe", -999)
    corr = r.get("corr_xs01", 1.0)
    distinct = corr is None or abs(corr) < 0.6
    return (int(earns), int(distinct and ho > 0 and full > 0), ho)

best = max(results, key=score_result)

print("\n" + "=" * 60)
print("BEST CONFIG:")
print(xs.fmt(best))
print("JSON:", xs.as_json(best))