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

"""MRV06 — VWAP Deviation Reversion

IDEA: On 1h bars, compute a rolling session VWAP (using typical price * volume).
Fade deviations > k*sigma back to VWAP (mean-reversion).
Regime gate: only trade in the direction of the daily trend (using a simple trend filter).

Variants tested:
  - k = 1.5 vs 2.0 (deviation threshold)
  - sigma window = 24h vs 48h (rolling window for sigma)

TF: 1h (VWAP is most meaningful at 1h granularity)
Style: continuous weights (study_weights)
"""

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


def compute_vwap_deviation(df: pd.DataFrame, vwap_win: int, k: float,
                           sigma_win: int) -> np.ndarray:
    """
    Compute VWAP deviation signal with regime gate.

    VWAP: rolling typical_price * volume / rolling volume (causal window).
    Signal: when price deviates > k*sigma above VWAP -> short (expect reversion)
            when price deviates > k*sigma below VWAP -> long (expect reversion)
    Regime gate: only long when daily trend (slow EMA > fast EMA at 1h scale).

    All computations causal (value at i uses data <= i).
    """
    close = df["close"].values.astype(float)
    high = df["high"].values.astype(float)
    low = df["low"].values.astype(float)
    volume = df["volume"].values.astype(float)

    # Typical price (causal: same bar is fine, we're using it for VWAP at i)
    typical = (high + low + close) / 3.0

    # Rolling VWAP (causal window)
    s = pd.Series
    tp_vol = typical * np.where(volume > 0, volume, np.nan)

    # Rolling VWAP over vwap_win bars
    vwap_num = s(tp_vol).rolling(vwap_win, min_periods=vwap_win // 2).sum()
    vwap_den = s(volume).rolling(vwap_win, min_periods=vwap_win // 2).sum()
    vwap = (vwap_num / vwap_den.replace(0, np.nan)).values

    # Deviation from VWAP
    deviation = close - vwap

    # Rolling sigma of deviation
    sigma = s(deviation).rolling(sigma_win, min_periods=sigma_win // 2).std().values

    # Normalized deviation (z-score wrt rolling sigma)
    z = np.where(sigma > 0, deviation / sigma, 0.0)

    # Mean-reversion signal:
    # z > k => price is too high above VWAP => short (negative position)
    # z < -k => price is too low below VWAP => long (positive position)
    # Gradual: use -z/k clipped to [-1, 1] when |z| > k, else 0
    signal = np.where(np.abs(z) > k, -np.sign(z), 0.0)

    # Regime gate using daily trend: EMA(50d) vs EMA(10d) at 1h scale
    # Only allow long when fast EMA > slow EMA (uptrend), allow short any time
    # (crypto is fundamentally bullish-biased; mean-reversion shorts in downtrend risky)
    ema_fast = al.ema(close, 10 * 24)  # 10-day EMA
    ema_slow = al.ema(close, 50 * 24)  # 50-day EMA

    # In uptrend (fast > slow): allow both long and short mean-reversion
    # In downtrend (fast < slow): allow only short mean-reversion (with VWAP)
    uptrend = ema_fast > ema_slow

    # Filter: only take longs in uptrend regime
    gated = np.where(signal > 0, signal * uptrend.astype(float), signal)

    # Apply vol-targeting for position sizing
    result = al.vol_target(gated, df, target_vol=0.20, vol_win_days=30, leverage_cap=2.0)
    result = np.nan_to_num(result, nan=0.0)

    return result


def make_target(vwap_win: int, k: float, sigma_win: int):
    """Factory: returns a target_fn(df) -> weights array."""
    def target_fn(df: pd.DataFrame) -> np.ndarray:
        return compute_vwap_deviation(df, vwap_win=vwap_win, k=k, sigma_win=sigma_win)
    target_fn.__name__ = f"vwap_dev_w{vwap_win}_k{k}_s{sigma_win}"
    return target_fn


# Small internal grid (<=4 param sets)
# VWAP window: 24h (1 session) vs 48h (2 sessions)
# k threshold: 1.5 vs 2.0
# sigma_win tied to vwap_win
CONFIGS = [
    # (vwap_win, k, sigma_win, label)
    (24,  1.5, 48,  "vwap24h_k1.5_s48h"),
    (24,  2.0, 48,  "vwap24h_k2.0_s48h"),
    (48,  1.5, 96,  "vwap48h_k1.5_s96h"),
    (48,  2.0, 96,  "vwap48h_k2.0_s96h"),
]

best_rep = None
best_hold = -999.0

print("=== MRV06 VWAP Deviation Reversion ===")
print(f"Testing {len(CONFIGS)} configs on 1h bars (BTC + ETH)\n")

for vwap_win, k, sigma_win, label in CONFIGS:
    print(f"--- Config: {label} ---")
    fn = make_target(vwap_win, k, sigma_win)
    rep = al.study_weights(
        f"MRV06-{label}",
        fn,
        tfs=("1h",)
    )
    print(al.fmt(rep))
    hold_sharpe = rep["verdict"].get("best_holdout_sharpe", -999)
    if hold_sharpe > best_hold:
        best_hold = hold_sharpe
        best_rep = rep
    print()

# Print best config
print("\n\n=== BEST CONFIG ===")
print(al.fmt(best_rep))
print("JSON:", al.as_json(best_rep))