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

"""XV05 — Low Max-Drawdown Anomaly
Score = -rolling_maxdrawdown(close, W) over the past W bars.
Prefer assets with smooth price history (low drawdown) for long,
prefer highly-drawn-down assets for short.

Grid: vary W (30, 60, 90), universe (majors, all), H (10).
<=5 study_xs calls total.
"""
import sys
sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/xsec")
import xslib as xs
import numpy as np


def rolling_maxdd(close, W):
    """Causal rolling max-drawdown over the past W bars.
    At row i: max drawdown of the window [i-W+1 .. i].
    Returns matrix (n_days x n_assets). NaN for first W-1 rows.
    Higher = worse drawdown (more negative).
    """
    n, A = close.shape
    out = np.full((n, A), np.nan)
    for i in range(W - 1, n):
        window = close[i - W + 1: i + 1]  # shape (W, A) — causal: data <= i
        # rolling peak up to each bar within window
        peak = np.maximum.accumulate(window, axis=0)
        dd = (window - peak) / peak  # drawdown at each bar (<=0)
        out[i] = np.nanmin(dd, axis=0)  # worst drawdown in window (most negative)
    return out


def score_fn_w60(P):
    """Score = -maxDD(W=60): prefer LOW drawdown (smooth equity)."""
    return -rolling_maxdd(P.close, 60)


def score_fn_w30(P):
    """Score = -maxDD(W=30): shorter memory."""
    return -rolling_maxdd(P.close, 30)


def score_fn_w90(P):
    """Score = -maxDD(W=90): longer memory."""
    return -rolling_maxdd(P.close, 90)


def score_fn_w60_blend(P):
    """Blend: average score from W=30 and W=90 (multi-horizon like XS01 blend)."""
    s30 = -rolling_maxdd(P.close, 30)
    s90 = -rolling_maxdd(P.close, 90)
    return xs.xs_zscore(s30) + xs.xs_zscore(s90)


if __name__ == "__main__":
    print("=== XV05: Low Max-Drawdown Anomaly ===\n")

    # Run 1: canonical W=60, majors universe, H=10, k=5, long-short
    print("--- Run 1: W=60, majors, H=10, k=5, LS ---")
    r1 = xs.study_xs("XV05-W60-maj", score_fn_w60, universe="majors", H=10, k=5, long_short=True)
    print(xs.fmt(r1))
    print()

    # Run 2: W=60, all universe (49 alts)
    print("--- Run 2: W=60, all, H=10, k=5, LS ---")
    r2 = xs.study_xs("XV05-W60-all", score_fn_w60, universe="all", H=10, k=5, long_short=True)
    print(xs.fmt(r2))
    print()

    # Run 3: W=30, majors
    print("--- Run 3: W=30, majors, H=10, k=5, LS ---")
    r3 = xs.study_xs("XV05-W30-maj", score_fn_w30, universe="majors", H=10, k=5, long_short=True)
    print(xs.fmt(r3))
    print()

    # Run 4: W=90, majors
    print("--- Run 4: W=90, majors, H=10, k=5, LS ---")
    r4 = xs.study_xs("XV05-W90-maj", score_fn_w90, universe="majors", H=10, k=5, long_short=True)
    print(xs.fmt(r4))
    print()

    # Run 5: blend W=30+W=90, all universe
    print("--- Run 5: Blend W30+W90, all, H=10, k=5, LS ---")
    r5 = xs.study_xs("XV05-BLENDall", score_fn_w60_blend, universe="all", H=10, k=5, long_short=True)
    print(xs.fmt(r5))
    print()

    # Pick best by earns_slot, then hold-out sharpe, then distinctness from XS01
    results = [r1, r2, r3, r4, r5]
    names = ["W60-maj", "W60-all", "W30-maj", "W90-maj", "Blend-all"]

    def score_key(r):
        earns = 1 if r["earns_slot"] else 0
        hold_sh = r["holdout"].get("sharpe", -99)
        full_sh = r["full"]["sharpe"]
        xs01_corr = abs(r["corr_xs01"] or 1.0)
        return (earns, hold_sh, full_sh, -xs01_corr)

    best = max(results, key=score_key)
    print("\n=== BEST CONFIG ===")
    print(xs.fmt(best))
    print("JSON:", xs.as_json(best))