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

"""MRV04 — IBS (Internal Bar Strength) Mean-Reversion

HYPOTHESIS: Internal Bar Strength = (close - low) / (high - low).
Long when IBS < low_thresh (closed near low = oversold position within bar),
flat (or short) when IBS > high_thresh (closed near high = overbought).

Classic daily mean-reversion edge. Testing on certified BTC/ETH daily bars.

Variants tested:
  V1: Long-flat  thresholds 0.20/0.80 (classic textbook)
  V2: Long-flat  thresholds 0.25/0.75 (slightly wider)
  V3: Long-short thresholds 0.20/0.80 (adds short leg)
  V4: Long-flat  thresholds 0.15/0.85 (tighter = rarer signals)
  Best variant selected by min-asset hold-out Sharpe.

All positions are vol-targeted (20% annualized, 2× leverage cap).
Evaluated on 1d timeframe (IBS is a daily signal by design).
"""

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


# ---------------------------------------------------------------------------
# IBS calculation (causal: uses close, high, low of the same bar i)
# ---------------------------------------------------------------------------
def ibs(df) -> np.ndarray:
    h = df["high"].values.astype(float)
    l = df["low"].values.astype(float)
    c = df["close"].values.astype(float)
    rng = h - l
    # Avoid division by zero (doji bars with zero range)
    result = np.where(rng > 0, (c - l) / rng, 0.5)
    return result


# ---------------------------------------------------------------------------
# Variant builders
# ---------------------------------------------------------------------------
def make_ibs_longflat(low_thresh: float, high_thresh: float):
    """Long when IBS < low_thresh, flat when IBS > high_thresh, hold otherwise."""
    def target_fn(df):
        ibs_val = ibs(df)
        pos = np.full(len(df), np.nan)
        pos[0] = 0.0
        for i in range(1, len(df)):
            if ibs_val[i] < low_thresh:
                pos[i] = 1.0   # go long
            elif ibs_val[i] > high_thresh:
                pos[i] = 0.0   # go flat
            else:
                pos[i] = pos[i - 1]  # hold
        pos = np.nan_to_num(pos, nan=0.0)
        return al.vol_target(pos, df, target_vol=0.20, vol_win_days=30, leverage_cap=2.0)
    return target_fn


def make_ibs_longshort(low_thresh: float, high_thresh: float):
    """Long when IBS < low_thresh, short when IBS > high_thresh, hold otherwise."""
    def target_fn(df):
        ibs_val = ibs(df)
        pos = np.full(len(df), np.nan)
        pos[0] = 0.0
        for i in range(1, len(df)):
            if ibs_val[i] < low_thresh:
                pos[i] = 1.0   # go long
            elif ibs_val[i] > high_thresh:
                pos[i] = -1.0  # go short
            else:
                pos[i] = pos[i - 1]  # hold
        pos = np.nan_to_num(pos, nan=0.0)
        return al.vol_target(pos, df, target_vol=0.20, vol_win_days=30, leverage_cap=2.0)
    return target_fn


# ---------------------------------------------------------------------------
# Vectorized version (faster, equivalent logic using ffill)
# ---------------------------------------------------------------------------
def make_ibs_longflat_vec(low_thresh: float, high_thresh: float):
    """Vectorized long-flat IBS strategy."""
    def target_fn(df):
        ibs_val = ibs(df)
        # Signal: 1=long, 0=flat, NaN=hold (ffill)
        sig = np.where(ibs_val < low_thresh, 1.0,
               np.where(ibs_val > high_thresh, 0.0, np.nan))
        sig[0] = 0.0  # start flat
        pos = sig.copy()
        # forward-fill NaN (hold previous)
        import pandas as pd
        pos = pd.Series(pos).ffill().fillna(0.0).values
        return al.vol_target(pos, df, target_vol=0.20, vol_win_days=30, leverage_cap=2.0)
    return target_fn


def make_ibs_longshort_vec(low_thresh: float, high_thresh: float):
    """Vectorized long-short IBS strategy."""
    def target_fn(df):
        import pandas as pd
        ibs_val = ibs(df)
        sig = np.where(ibs_val < low_thresh, 1.0,
               np.where(ibs_val > high_thresh, -1.0, np.nan))
        sig[0] = 0.0
        pos = pd.Series(sig).ffill().fillna(0.0).values
        return al.vol_target(pos, df, target_vol=0.20, vol_win_days=30, leverage_cap=2.0)
    return target_fn


# ---------------------------------------------------------------------------
# Run all variants
# ---------------------------------------------------------------------------
if __name__ == "__main__":
    TFS = ("1d",)

    variants = [
        ("MRV04-V1-LF-0.20/0.80", make_ibs_longflat_vec(0.20, 0.80)),
        ("MRV04-V2-LF-0.25/0.75", make_ibs_longflat_vec(0.25, 0.75)),
        ("MRV04-V3-LS-0.20/0.80", make_ibs_longshort_vec(0.20, 0.80)),
        ("MRV04-V4-LF-0.15/0.85", make_ibs_longflat_vec(0.15, 0.85)),
    ]

    results = []
    for name, fn in variants:
        print(f"\nRunning {name} ...")
        rep = al.study_weights(name, fn, tfs=TFS)
        print(al.fmt(rep))
        results.append(rep)

    # Pick best by min_asset_holdout_sharpe
    best = max(results, key=lambda r: r["verdict"].get("best_holdout_sharpe", -99))
    print("\n" + "=" * 60)
    print(f"BEST VARIANT: {best['name']}")
    print(al.fmt(best))
    print("JSON:", al.as_json(best))