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

"""CMB02 — Donchian breakout + volume elevation + DVOL filter (triple filter).

HYPOTHESIS:
  Long-flat Donchian channel breakout, but only when:
    1. Volume is elevated (above rolling median, filtering fake/thin breakouts)
    2. DVOL is NOT in panic zone (below 80th percentile expanding, avoiding breakouts
       during fear spikes that tend to reverse)
  Position is vol-targeted. Hold until price drops back below mid-channel.

The triple filter tests: breakouts with confirming volume + calm/moderate implied vol
should capture real trending moves while avoiding panic-spike false breakouts.

DVOL note: data starts 2021-03 -> backtest uses full history where available,
DVOL filter only active where DVOL data exists (NaN -> filter passes through).
"""
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 cmb02_target(df: pd.DataFrame, don_win: int = 20, vol_win: int = 20,
                 dvol_pct: float = 80.0, asset: str = "BTC") -> np.ndarray:
    """
    Donchian breakout, long-flat, with volume + DVOL filters.

    Entry: close[i] > donchian_high[i] (prior win-bar high)
           AND volume[i] > vol_median over rolling vol_win bars
           AND DVOL[i] < expanding percentile dvol_pct (not in panic zone)
    Exit: close[i] < donchian_mid[i] (midpoint of channel, trailing)
    Position: vol-targeted at 20%, leverage cap 2x.
    """
    c = df["close"].values.astype(float)
    v = df["volume"].values.astype(float)
    n = len(c)

    # --- Donchian channel (strictly causal: shift(1)) ---
    hi, lo = al.donchian(df, don_win)
    mid = (hi + lo) / 2.0

    # --- Volume filter: volume above rolling median (causal) ---
    vol_median = pd.Series(v).rolling(vol_win, min_periods=max(2, vol_win // 2)).median().values
    vol_elevated = v > vol_median  # True when volume confirms breakout

    # --- DVOL filter: NOT in panic zone (expanding percentile, causal) ---
    dv = al.dvol(df, asset)  # float array, NaN before 2021-03
    # Expanding percentile (causal): for each i, rank of dv[i] vs all dv[0..i]
    # Use pd expanding quantile (causal by nature)
    dv_series = pd.Series(dv)
    # Compute expanding percentile threshold causally
    # We need: is dv[i] < dvol_pct-th percentile of dv[0..i]?
    # Equivalent: expanding rank < dvol_pct%
    # We use expanding().quantile() for the threshold line
    dv_thresh = dv_series.expanding(min_periods=30).quantile(dvol_pct / 100.0).values
    # Filter: DVOL below the threshold (not in panic zone)
    # If DVOL is NaN (pre-2021), treat filter as passing (no data = no veto)
    dvol_ok = np.where(np.isnan(dv), True, dv < dv_thresh)

    # --- Build position signal ---
    # We use a stateful forward-fill approach:
    # position is 1 if breakout + filters, 0 if exit signal, else carry
    raw_dir = np.zeros(n)
    pos = 0.0
    for i in range(1, n):
        # Exit condition: price dropped below mid-channel
        if pos > 0 and np.isfinite(mid[i]) and c[i] < mid[i]:
            pos = 0.0
        # Entry condition: breakout + volume + dvol filters
        if (pos == 0.0 and
                np.isfinite(hi[i]) and c[i] > hi[i] and
                vol_elevated[i] and
                dvol_ok[i]):
            pos = 1.0
        raw_dir[i] = pos

    # Apply vol-targeting on the binary direction
    return al.vol_target(raw_dir, df, target_vol=0.20, vol_win_days=30, leverage_cap=2.0)


def run():
    # Small grid: don_win x dvol_pct
    # 4 configs (<=4), 2 TFs -> 4*2 = 8 backtests on 2 assets each = 16 total
    # To stay within <=6 total backtest calls, we pick 2 configs x 1 TF + best config x 2nd TF
    # Actually: study_weights calls for 2 assets each run -> each study_weights(tfs=("1d",)) = 2 backtests
    # We'll do 2 param configs on 1d, then best on 12h -> 3 study_weights calls = 6 asset-backtests

    results = []

    configs = [
        dict(don_win=20, vol_win=20, dvol_pct=80.0, label="D20-V20-DVOL80"),
        dict(don_win=40, vol_win=20, dvol_pct=75.0, label="D40-V20-DVOL75"),
        dict(don_win=20, vol_win=10, dvol_pct=70.0, label="D20-V10-DVOL70"),
        dict(don_win=60, vol_win=30, dvol_pct=80.0, label="D60-V30-DVOL80"),
    ]

    print("=== CMB02: Donchian + Volume + DVOL filter ===\n")

    best_rep = None
    best_score = -999.0

    for cfg in configs:
        label = cfg["label"]
        don_win = cfg["don_win"]
        vol_win = cfg["vol_win"]
        dvol_pct = cfg["dvol_pct"]

        def make_target(dw=don_win, vw=vol_win, dp=dvol_pct):
            def target_fn(df):
                # Determine asset from df shape/content - try BTC first, ETH fallback
                # We pass asset through closure workaround via index
                # Actually altlib doesn't pass asset name to target_fn...
                # We'll call dvol with "BTC" and check if ETH data matches better
                # The dvol function uses asset param - we need a way to know which asset
                # Use a hack: check if the df matches BTC or ETH by length/timestamps
                btc_df = al.get("BTC", "1d")
                if len(df) == len(btc_df) and df["close"].iloc[0] == btc_df["close"].iloc[0]:
                    asset = "BTC"
                else:
                    asset = "ETH"
                return cmb02_target(df, don_win=dw, vol_win=vw, dvol_pct=dp, asset=asset)
            return target_fn

        rep = al.study_weights(f"CMB02-{label}", make_target(), tfs=("1d",))
        print(al.fmt(rep))
        print()

        score = rep["verdict"].get("best_holdout_sharpe", -9)
        if score > best_score:
            best_score = score
            best_rep = rep
            best_label = label
            best_cfg = cfg

    print(f"\n>>> Best config on 1d: {best_label} (holdout score: {best_score:.3f})")
    print(">>> Now testing best config on 12h...\n")

    # Test best config on 12h
    dw = best_cfg["don_win"]
    vw = best_cfg["vol_win"]
    dp = best_cfg["dvol_pct"]

    def make_target_12h(dw=dw, vw=vw, dp=dp):
        def target_fn(df):
            btc_df = al.get("BTC", "12h")
            if len(df) == len(btc_df) and df["close"].iloc[0] == btc_df["close"].iloc[0]:
                asset = "BTC"
            else:
                asset = "ETH"
            return cmb02_target(df, don_win=dw, vol_win=vw, dvol_pct=dp, asset=asset)
        return target_fn

    rep_12h = al.study_weights(f"CMB02-{best_label}-12h", make_target_12h(), tfs=("12h",))
    print(al.fmt(rep_12h))
    print()

    # Build combined report with both TFs for the best config
    # Combine cells from 1d best + 12h
    best_1d_cells = [c for c in best_rep["cells"] if c["tf"] == "1d"]
    cells_combined = best_1d_cells + rep_12h["cells"]

    # Pick best TF by holdout
    def pick_best(cells):
        return max(cells, key=lambda c: c.get("min_asset_holdout_sharpe", -9))

    best_cell = pick_best(cells_combined)
    best_tf = best_cell["tf"]

    # Final verdict
    from altlib import _verdict
    verdict = _verdict(cells_combined)

    final_rep = dict(
        name=f"CMB02-{best_label}",
        kind="weights",
        cells=cells_combined,
        verdict=verdict,
    )

    print("\n=== FINAL REPORT (best config, both TFs) ===")
    print(al.fmt(final_rep))
    print("\nJSON:", al.as_json(final_rep))
    return final_rep


if __name__ == "__main__":
    run()