PythagorasGoal/scripts/research/alt/runs/RSK01.py

"""RSK01 — Vol-target B&H + DD breaker.

Hypothesis: Long-only vol-targeted (no trend signal) with a circuit breaker:
  - Normally always long, scaled by vol-targeting (target 20%, cap 2x)
  - Goes FLAT when the strategy equity drawdown from peak exceeds `dd_thresh`
  - Re-enters when the MARKET (asset price) recovers by `recovery_frac` from its
    trough level at the time the breaker fired
    (NOTE: recovery on MARKET price, not strategy equity — otherwise the flat
     position freezes equity and the breaker never clears, a death spiral)
  - Does the breaker beat pure vol-targeted buy&hold?

Grid: 3 configs on 1d (= 6 asset-backtests, within the 6-backtest limit).
"""

import sys
sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/alt")
import altlib as al
import numpy as np


def rsk01_target(df, dd_thresh: float = 0.15, recovery_frac: float = 0.50) -> np.ndarray:
    """
    Causal vol-targeted long-only position with equity-DD circuit breaker.

    Breaker fires when strategy equity drawdown > dd_thresh.
    Recovery: re-enter when asset price has risen by recovery_frac * (asset price drop
    from the time breaker fired). This is observable from MARKET price, avoids death-spiral.

    At each bar i:
      1. Base vol-targeted position (direction=+1) computed causally
      2. Simulated strategy equity updated by previous bar's held position
      3. If equity-DD > dd_thresh → BREAKER ON, record price_trough = close[i]
      4. BREAKER recovers when close[i] >= price_trough * (1 + recovery_frac * rel_drop)
         where rel_drop = (price_at_breaker_on - price_trough_at_bar_i) / price_at_breaker_on
         More simply: re-enter when close[i] >= price_trough * (1 + recovery_frac * dd_thresh)
    """
    c = df["close"].values.astype(float)
    r = al.simple_returns(c)

    # Base vol-targeted position (always long direction=+1)
    direction = np.ones(len(c))
    base_pos = al.vol_target(direction, df, target_vol=0.20, vol_win_days=30, leverage_cap=2.0)

    n = len(c)
    final_pos = np.zeros(n)

    # Strategy equity tracking (causal: equity at i reflects positions through i-1)
    eq = 1.0
    peak = 1.0
    breaker_on = False
    price_trough = np.nan    # asset price when breaker fired
    recovery_target_price = np.nan  # asset price target for re-entry

    for i in range(n):
        # Update strategy equity from previous bar's position
        if i > 0:
            prev_pos = final_pos[i - 1]
            eq *= (1.0 + prev_pos * r[i])

        # Update running equity peak
        if eq > peak:
            peak = eq

        dd = (peak - eq) / peak if peak > 0 else 0.0
        price_now = c[i]

        if not breaker_on:
            if dd > dd_thresh:
                breaker_on = True
                # Record asset price trough at breakout trigger
                price_trough = price_now
                # Recovery target: price rises by recovery_frac * dd_thresh above trough
                # (dd_thresh is a proxy for the % drop in the asset that caused the DD)
                recovery_target_price = price_trough * (1.0 + recovery_frac * dd_thresh)
        else:
            # Re-enter when asset recovers to recovery_target_price
            if price_now >= recovery_target_price:
                breaker_on = False
                price_trough = np.nan
                recovery_target_price = np.nan
                # Also reset the equity peak to current level to avoid immediate re-trigger
                peak = eq

        final_pos[i] = 0.0 if breaker_on else base_pos[i]

    return final_pos


def make_target(dd_thresh: float, recovery_frac: float):
    """Factory to create a target function with fixed params."""
    def _target(df):
        return rsk01_target(df, dd_thresh=dd_thresh, recovery_frac=recovery_frac)
    _target.__name__ = f"RSK01_dd{int(dd_thresh*100)}_rec{int(recovery_frac*100)}"
    return _target


# Grid: 3 configs on 1d (= 6 asset-backtests, within the 6-backtest limit)
CONFIGS_SCREEN = [
    (0.10, 0.50),   # tight breaker, recover 50% of dd_thresh in price terms
    (0.15, 0.50),   # moderate breaker
    (0.20, 0.50),   # loose breaker
]

print("=== RSK01: Vol-target B&H + DD circuit breaker ===")
print("Recovery measured on MARKET PRICE (not frozen strategy equity)")
print("Screening 3 configs on 1d (6 asset-backtests)...")
print()

best_rep = None
best_score = -999
best_cfg = None

for dd_thresh, rec_frac in CONFIGS_SCREEN:
    name = f"RSK01_dd{int(dd_thresh*100)}_rec{int(rec_frac*100)}"
    target_fn = make_target(dd_thresh, rec_frac)
    rep = al.study_weights(name, target_fn, tfs=("1d",))

    score = rep["verdict"].get("best_holdout_sharpe", -9)
    btc = rep["cells"][0]["per_asset"]["BTC"]
    eth = rep["cells"][0]["per_asset"]["ETH"]
    print(f"  {name}:")
    print(f"    BTC: full Sh={btc['full']['sharpe']:.2f} DD={btc['full']['maxdd']:.1%} "
          f"TIM={btc['tim']:.1%} hold Sh={btc['holdout']['sharpe']:.2f}")
    print(f"    ETH: full Sh={eth['full']['sharpe']:.2f} DD={eth['full']['maxdd']:.1%} "
          f"TIM={eth['tim']:.1%} hold Sh={eth['holdout']['sharpe']:.2f}")
    print(f"    grade={rep['verdict']['grade']} minFull={rep['verdict'].get('best_full_sharpe'):.2f} "
          f"minHold={score:.2f}")
    print()

    if score > best_score:
        best_score = score
        best_rep = rep
        best_cfg = (dd_thresh, rec_frac)

print(f"Best config: dd_thresh={best_cfg[0]}, recovery_frac={best_cfg[1]}")
print()

# Final clean report on best config
dd_thresh, rec_frac = best_cfg
name = f"RSK01_dd{int(dd_thresh*100)}_rec{int(rec_frac*100)}"
target_fn = make_target(dd_thresh, rec_frac)
final_rep = al.study_weights(name, target_fn, tfs=("1d",))

print(al.fmt(final_rep))
print("JSON:", al.as_json(final_rep))