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

"""XR02 — Short-term Reversal gated by high-vol regime (L=3).

MECHANISM:
  Plain reversal: short the recent winners, long the recent losers (L=3 days lookback).
  GATE: only active when market volatility is HIGH — defined as the 30-day rolling std of
  the equal-weight market return exceeding its own 90-day expanding percentile (p70 threshold).
  In low-vol / calm regimes, the book is flat (score = NaN -> no position).

Rationale: short-term reversal is a classic effect but is often diluted by trend in calm
regimes. In panic / high-vol regimes (sharp market moves), mean-reversion / liquidity
provision logic is stronger (overshoot + reversal). Gate concentrates the signal in those
regimes while avoiding trend-contamination in smooth uptrends.

Causal: all quantities computed at close[i], applied to return of bar i+1.
"""

import sys
sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/xsec")
import xslib as xs
import numpy as np
import pandas as pd


def rev_gate_score(P, L=3, vol_win=30, baseline_win=90, vol_pct=70, use_residual=False):
    """Short-term reversal score gated by high market-vol regime.

    Parameters
    ----------
    P : Panel
    L : int
        Reversal lookback in days (price L days ago vs today).
    vol_win : int
        Rolling window for realised market-vol (std of equal-weight market return).
    baseline_win : int
        Expanding window for computing the percentile threshold on market-vol.
    vol_pct : float
        Percentile threshold: market vol must exceed this percentile to be active.
    use_residual : bool
        If True, compute reversal on idiosyncratic (market-beta-neutral) returns instead of raw.
    """
    n, A = P.close.shape

    # --- 1. Reversal score: negative of L-day return (higher = long the loser) ---
    raw_ret = xs.past_return(P.close, L)   # (n, A), causal: uses close[i-L..i]
    if use_residual:
        # use idiosyncratic cumulative return instead of total
        resid = xs.residual_return(P.ret, 30)   # (n, A), causal
        # cumulate idiosyncratic over L days
        resid_cum = np.full_like(raw_ret, np.nan)
        for lag in range(1, L + 1):
            shifted = np.roll(resid, lag, axis=0)
            shifted[:lag] = np.nan
            resid_cum = np.nansum([resid_cum, resid], axis=0)
        signal = -resid_cum
    else:
        signal = -raw_ret   # reversal: short winners, long losers

    # --- 2. Market-vol regime gate (expanding percentile, causal) ---
    mkt = xs.market_ret(P.ret)                               # (n,) equal-weight mkt return
    mkt_vol = pd.Series(mkt).rolling(vol_win, min_periods=max(5, vol_win // 2)).std().values
    # expanding percentile of mkt_vol up to each row i (causal)
    thresh = np.full(n, np.nan)
    for i in range(baseline_win, n):
        hist = mkt_vol[max(0, i - baseline_win):i + 1]
        hist = hist[np.isfinite(hist)]
        if len(hist) >= 10:
            thresh[i] = np.nanpercentile(hist, vol_pct)

    # gate: active only when mkt_vol > threshold (high-vol regime)
    active = (mkt_vol > thresh)   # (n,) boolean, NaN -> False
    active[~np.isfinite(mkt_vol) | ~np.isfinite(thresh)] = False

    # --- 3. Apply gate: set score to NaN when flat ---
    score = signal.copy()
    score[~active, :] = np.nan
    return score


# ---------------------------------------------------------------------------
# GRID — <=5 study_xs calls
# Config space: L in {3, 5}, vol_pct in {60, 70}, universe in {majors, all}
# ---------------------------------------------------------------------------

print("=" * 70)
print("XR02: Short-term Reversal gated by high-vol regime")
print("=" * 70)

results = []

# Config 1: L=3, pct=70, majors (baseline config)
print("\n[1/5] L=3, vol_pct=70, universe=majors")
rep1 = xs.study_xs(
    "XR02-L3-p70-maj",
    lambda P: rev_gate_score(P, L=3, vol_win=30, baseline_win=90, vol_pct=70),
    universe="majors",
    H=3,
    k=5,
    long_short=True,
)
print(xs.fmt(rep1))
results.append(rep1)

# Config 2: L=3, pct=70, all (wider universe)
print("\n[2/5] L=3, vol_pct=70, universe=all")
rep2 = xs.study_xs(
    "XR02-L3-p70-all",
    lambda P: rev_gate_score(P, L=3, vol_win=30, baseline_win=90, vol_pct=70),
    universe="all",
    H=3,
    k=5,
    long_short=True,
)
print(xs.fmt(rep2))
results.append(rep2)

# Config 3: L=3, pct=60 (more permissive gate), majors
print("\n[3/5] L=3, vol_pct=60, universe=majors")
rep3 = xs.study_xs(
    "XR02-L3-p60-maj",
    lambda P: rev_gate_score(P, L=3, vol_win=30, baseline_win=90, vol_pct=60),
    universe="majors",
    H=3,
    k=5,
    long_short=True,
)
print(xs.fmt(rep3))
results.append(rep3)

# Config 4: L=5, pct=70, majors (longer lookback)
print("\n[4/5] L=5, vol_pct=70, universe=majors")
rep4 = xs.study_xs(
    "XR02-L5-p70-maj",
    lambda P: rev_gate_score(P, L=5, vol_win=30, baseline_win=90, vol_pct=70),
    universe="majors",
    H=5,
    k=5,
    long_short=True,
)
print(xs.fmt(rep4))
results.append(rep4)

# Config 5: L=3, pct=70, majors, H=5 (slower rebalance)
print("\n[5/5] L=3, vol_pct=70, universe=majors, H=5")
rep5 = xs.study_xs(
    "XR02-L3-p70-maj-H5",
    lambda P: rev_gate_score(P, L=3, vol_win=30, baseline_win=90, vol_pct=70),
    universe="majors",
    H=5,
    k=5,
    long_short=True,
)
print(xs.fmt(rep5))
results.append(rep5)

# ---------------------------------------------------------------------------
# BEST: pick by earns_slot, then holdout sharpe, then distinctness
# ---------------------------------------------------------------------------
def score_result(r):
    earns = int(r.get("earns_slot", False))
    hold_sh = r["holdout"].get("sharpe", -99)
    full_sh = r["full"].get("sharpe", -99)
    corr_xs01 = r.get("corr_xs01") or 1.0
    # prefer earns_slot, then hold-out, then distinctness, then full
    return (earns, hold_sh, full_sh, -abs(corr_xs01))

best = max(results, key=score_result)

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