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

"""XAS01 — ETH/BTC ratio z-score reversion strategy.

IDEA: The ETH/BTC ratio (price ratio) exhibits mean-reversion. When the z-score of the
ratio falls below a threshold (ETH is cheap relative to BTC), go long the ratio
(long ETH, short BTC). When z-score rises above threshold, go short the ratio.

IMPLEMENTATION:
- Build ratio = ETH_close / BTC_close on aligned timestamps (inner join).
- Compute rolling z-score of the log-ratio over a lookback window.
- Position: +1 when z < -threshold (long ratio), -1 when z > +threshold (short ratio), 0 otherwise.
- The SPREAD P&L is: pos * (ETH_return - BTC_return) per bar.
- We use eval_weights on a synthetic series where close = ratio, so that simple_returns(ratio)
  gives the ratio return which equals ETH_return - BTC_return (approximately for log returns).
- Actually: ratio_return = ETH/BTC new / ETH/BTC old - 1 ≈ r_ETH - r_BTC (log approximation)
  But for precise spread return: r_spread = r_ETH - r_BTC exactly in log space.
  We construct a synthetic df with close=ratio so eval_weights gives us ratio simple returns.

GRID (4 configs, 2 TFs = 8 backtests — within limit):
  lookback_days: [20, 60]
  threshold: [1.5, 2.0]

We pick the best config (highest min holdout sharpe) and report that.
"""

import sys
import json
import numpy as np
import pandas as pd

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

# ===========================================================================
# Helper: build synthetic ratio df aligned between BTC and ETH
# ===========================================================================

def build_ratio_df(tf: str) -> pd.DataFrame:
    """Merge BTC and ETH on timestamp (inner join), build close=ETH/BTC ratio."""
    btc = al.get("BTC", tf)[["timestamp", "datetime", "close"]].rename(columns={"close": "btc"})
    eth = al.get("ETH", tf)[["timestamp", "close"]].rename(columns={"close": "eth"})
    merged = pd.merge(btc, eth, on="timestamp", how="inner").reset_index(drop=True)
    merged["close"] = merged["eth"] / merged["btc"]
    # Add stub OHLCV columns so eval_weights works (it only needs close)
    merged["open"] = merged["close"]
    merged["high"] = merged["close"]
    merged["low"] = merged["close"]
    merged["volume"] = 1.0
    return merged[["timestamp", "datetime", "open", "high", "low", "close", "volume"]]


# ===========================================================================
# Strategy: z-reversion on log(ratio)
# ===========================================================================

def make_target(lookback_days: int, threshold: float, vol_tgt: bool = True):
    """Return a target_fn(df) for eval_weights on the ratio df."""
    def target_fn(df: pd.DataFrame) -> np.ndarray:
        c = df["close"].values.astype(float)
        log_ratio = np.log(c)  # log(ETH/BTC)

        bpd = al.bars_per_day(df)
        win = max(2, lookback_days * bpd)

        z = al.zscore(log_ratio, win)

        # Mean-reversion: short when z > threshold (ratio overbought), long when z < -threshold
        direction = np.where(z < -threshold, 1.0,
                    np.where(z > threshold, -1.0, 0.0))

        if vol_tgt:
            # Vol-target the spread position
            pos = al.vol_target(direction, df, target_vol=0.20, vol_win_days=30, leverage_cap=2.0)
        else:
            pos = direction.astype(float)

        pos = np.nan_to_num(pos, nan=0.0)
        return pos

    return target_fn


# ===========================================================================
# Run grid: 2 lookbacks x 2 thresholds = 4 configs; 2 TFs = 8 backtests
# ===========================================================================

GRID = [
    {"lookback_days": 20, "threshold": 1.5},
    {"lookback_days": 20, "threshold": 2.0},
    {"lookback_days": 60, "threshold": 1.5},
    {"lookback_days": 60, "threshold": 2.0},
]

TFS = ("1d", "12h")

best_rep = None
best_score = -999.0
best_label = ""

print("=== XAS01: ETH/BTC ratio z-reversion ===")
print(f"Grid: {len(GRID)} configs x {len(TFS)} TFs = {len(GRID)*len(TFS)} backtests")
print()

for params in GRID:
    lb = params["lookback_days"]
    thr = params["threshold"]
    name = f"XAS01-lb{lb}-thr{thr}"
    print(f"--- {name} ---")

    # We need a custom study_weights that uses ratio df instead of per-asset dfs
    # Build ratio df for each TF, run eval_weights on it
    cells = []
    for tf in TFS:
        try:
            ratio_df = build_ratio_df(tf)
            tgt_fn = make_target(lb, thr, vol_tgt=True)
            tgt = tgt_fn(ratio_df)

            # Eval with fee sweep
            base = al.eval_weights(ratio_df, tgt, fee_side=al.FEE_SIDE)
            sweep = {f"{2*f*100:.2f}%RT": al.eval_weights(ratio_df, tgt, fee_side=f)["full"]["sharpe"]
                     for f in al.FEE_SWEEP}
            fee_ok = sweep.get("0.20%RT", -9) > 0

            full = base["full"]
            hold = base["holdout"]
            yearly = base["yearly"]

            print(f"  TF={tf}: full Sh={full['sharpe']:+.3f}, DD={full['maxdd']*100:.1f}%,"
                  f" hold Sh={hold.get('sharpe', 0):+.3f}, feeOK={fee_ok}")
            print(f"    fee sweep: {sweep}")

            cells.append(dict(
                tf=tf,
                per_asset={"RATIO": dict(full=full, holdout=hold, tim=base["time_in_market"],
                                         turnover=base["turnover_per_year"], fee_sweep=sweep,
                                         yearly=yearly)},
                min_asset_full_sharpe=round(full["sharpe"], 3),
                min_asset_holdout_sharpe=round(hold.get("sharpe", 0.0), 3),
                full_sharpe=round(full["sharpe"], 3),
                fee_survives=fee_ok,
            ))
        except Exception as e:
            print(f"  TF={tf}: ERROR {e}")

    # Score = best holdout sharpe across TFs
    score = max((c["min_asset_holdout_sharpe"] for c in cells), default=-999.0)
    if score > best_score:
        best_score = score
        best_label = name
        # Build a "rep" compatible with al.fmt
        # We adapt it to show ratio as both BTC and ETH (same series)
        adapted_cells = []
        for c in cells:
            ratio_res = c["per_asset"]["RATIO"]
            adapted_cells.append(dict(
                tf=c["tf"],
                per_asset={
                    "BTC": ratio_res,  # spread result attributed to both
                    "ETH": ratio_res,
                },
                min_asset_full_sharpe=c["min_asset_full_sharpe"],
                min_asset_holdout_sharpe=c["min_asset_holdout_sharpe"],
                full_sharpe=c["full_sharpe"],
                fee_survives=c["fee_survives"],
            ))
        best_rep = dict(name=name, kind="weights", cells=adapted_cells,
                        verdict=al._verdict(adapted_cells))
    print()

print()
print("=== BEST CONFIG:", best_label, "===")
print(al.fmt(best_rep))
print()
print("JSON:", al.as_json(best_rep))