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

"""XL04 [LIQ] — Dollar-volume momentum.

Score = past_return of dollar-volume (close * volume) over W=30 days.
Idea: assets gaining LIQUIDITY / ATTENTION relative to peers will outperform.
This is the OPPOSITE of XL03 (which went long LOW dollar-volume names).

Mechanism:
  dvol[i] = close[i] * vol[i]          (daily dollar volume)
  score[i] = dvol[i] / dvol[i-W] - 1  (W-day return of dollar volume)
  -> long assets whose dollar volume is GROWING the fastest

Grid (<=5 runs):
  1. baseline: universe=all, H=10, k=5, long_short=True, W=30
  2. shorter window W=10 (faster attention signal)
  3. longer window W=60 (more stable)
  4. majors universe (19 XS01 assets — check distinctness from XS01)
  5. long-only version (long attention gainers, no shorting attention losers)
"""
import sys
sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/xsec")
import xslib as xs
import numpy as np


# --- score factory -----------------------------------------------------------

def dvol_momentum_score(P, W=30):
    """Score = W-day past return of dollar volume (close * volume).
    CAUSAL: dvol_return[i] uses dvol[i] / dvol[i-W] - 1.
    Higher score = dollar volume growing faster = LONG.
    """
    dvol = P.close * P.vol           # (n, A) daily dollar volume
    score = np.full_like(dvol, np.nan)
    # past_return style: score[i] = dvol[i] / dvol[i-W] - 1
    # guard: if dvol[i-W] == 0 -> NaN
    denom = dvol[:-W]                # dvol[i-W]
    numer = dvol[W:]                 # dvol[i]
    with np.errstate(invalid="ignore", divide="ignore"):
        ratio = np.where(denom > 0, numer / denom - 1.0, np.nan)
    score[W:] = ratio
    return score


# --- grid -------------------------------------------------------------------

print("=" * 70)
print("XL04 [LIQ] Dollar-volume momentum — grid search")
print("=" * 70)

results = []

# Run 1: baseline (all, H=10, k=5, LS, W=30)
r1 = xs.study_xs("XL04-W30-all-LS",
                 lambda P: dvol_momentum_score(P, 30),
                 universe="all", H=10, k=5, long_short=True)
print(xs.fmt(r1))
print("JSON:", xs.as_json(r1))
results.append(r1)

# Run 2: shorter window W=10 (faster attention surge)
r2 = xs.study_xs("XL04-W10-all-LS",
                 lambda P: dvol_momentum_score(P, 10),
                 universe="all", H=10, k=5, long_short=True)
print(xs.fmt(r2))
print("JSON:", xs.as_json(r2))
results.append(r2)

# Run 3: longer window W=60 (sustained attention)
r3 = xs.study_xs("XL04-W60-all-LS",
                 lambda P: dvol_momentum_score(P, 60),
                 universe="all", H=10, k=5, long_short=True)
print(xs.fmt(r3))
print("JSON:", xs.as_json(r3))
results.append(r3)

# Run 4: majors universe only (19 XS01 assets)
r4 = xs.study_xs("XL04-W30-majors-LS",
                 lambda P: dvol_momentum_score(P, 30),
                 universe="majors", H=10, k=5, long_short=True)
print(xs.fmt(r4))
print("JSON:", xs.as_json(r4))
results.append(r4)

# Run 5: long-only (attention gainers only, no shorting losers)
r5 = xs.study_xs("XL04-W30-all-LO",
                 lambda P: dvol_momentum_score(P, 30),
                 universe="all", H=10, k=5, long_short=False)
print(xs.fmt(r5))
print("JSON:", xs.as_json(r5))
results.append(r5)

# --- pick best config -------------------------------------------------------
print("\n" + "=" * 70)
print("SUMMARY — best by: earns_slot first, then hold-out Sharpe, then distinctness from XS01")
print("=" * 70)


def rank_key(r):
    earns = int(r["earns_slot"])
    hold_sh = r["holdout"].get("sharpe", -9) or -9
    xs01_corr = abs(r.get("corr_xs01") or 1.0)
    full_sh = r["full"].get("sharpe", -9) or -9
    return (earns, hold_sh, full_sh, -xs01_corr)


best = max(results, key=rank_key)
print(f"\nBEST CONFIG: {best['name']}")
print(xs.fmt(best))
print("\nJSON (best):", xs.as_json(best))