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

"""XAS09 — Dual-momentum BTC/ETH
HYPOTHESIS: Absolute+relative momentum — hold the stronger asset only if its
absolute momentum > 0, else flat. Vol-targeted.

Logic (per bar i, decided with close[i], held during bar i+1):
  - Compute absolute momentum for BTC and ETH over lookback window L
    abs_mom_BTC = close_BTC[i] / close_BTC[i-L] - 1
    abs_mom_ETH = close_ETH[i] / close_ETH[i-L] - 1
  - Pick the stronger asset: whichever has higher momentum
  - Apply absolute-momentum gate: if the WINNER's abs_mom <= 0, go flat
  - Assign vol-targeted position (+1 or 0) to the winning asset; other = 0

This is a CROSS-ASSET strategy: the target for each asset depends on BOTH
BTC and ETH data aligned at the same bar. We align on datetime intersection.

Implementation as study_weights: called once per asset, so we need to align
the two series internally inside target_fn via global shared dfs.

We try a small grid of lookback windows: 1m (21d), 3m (63d), 6m (126d).
TFs: 1d and 12h — 3 lookbacks x 2 TFs = 6 backtests (within limit).
"""

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

# ---------------------------------------------------------------------------
# Helper: build dual-momentum targets for BOTH assets at once for a given TF.
# Returns (target_btc, target_eth) numpy arrays of equal length, aligned.
# ---------------------------------------------------------------------------

def dual_momentum_targets(tf: str, lookback_days: int,
                           target_vol: float = 0.20,
                           leverage_cap: float = 2.0):
    """
    Compute vol-targeted dual-momentum positions for BTC and ETH.

    Steps (all causal):
      1. Load BTC and ETH DataFrames for the given TF.
      2. Align on common datetime index (inner join).
      3. For each bar i:
           abs_mom_btc = close_btc[i] / close_btc[i-L] - 1
           abs_mom_eth = close_eth[i] / close_eth[i-L] - 1
           winner = asset with higher abs_mom
           gate   = winner's abs_mom > 0
           dir_btc = +1 if winner==BTC and gate else 0
           dir_eth = +1 if winner==ETH and gate else 0
      4. Vol-target each direction using that asset's own returns.
    """
    df_btc = al.get("BTC", tf).copy()
    df_eth = al.get("ETH", tf).copy()

    # Align on datetime (both from same source so should match; but be safe)
    df_btc["datetime"] = pd.to_datetime(df_btc["datetime"], utc=True)
    df_eth["datetime"] = pd.to_datetime(df_eth["datetime"], utc=True)

    # Merge on datetime (inner join = common bars only)
    merged = pd.merge(
        df_btc[["datetime", "close"]].rename(columns={"close": "close_btc"}),
        df_eth[["datetime", "close"]].rename(columns={"close": "close_eth"}),
        on="datetime", how="inner"
    ).reset_index(drop=True)

    n = len(merged)
    bpd = al.bars_per_day(df_btc)
    L = max(2, round(lookback_days * bpd))

    c_btc = merged["close_btc"].values.astype(float)
    c_eth = merged["close_eth"].values.astype(float)

    # Absolute momentum (causal: uses close[i] vs close[i-L])
    abs_mom_btc = np.full(n, np.nan)
    abs_mom_eth = np.full(n, np.nan)
    abs_mom_btc[L:] = c_btc[L:] / c_btc[:-L] - 1.0
    abs_mom_eth[L:] = c_eth[L:] / c_eth[:-L] - 1.0

    # Direction: +1 for winner asset, 0 for loser and flat periods
    dir_btc = np.zeros(n, dtype=float)
    dir_eth = np.zeros(n, dtype=float)

    valid = np.isfinite(abs_mom_btc) & np.isfinite(abs_mom_eth)

    # BTC stronger AND positive
    btc_wins = valid & (abs_mom_btc >= abs_mom_eth) & (abs_mom_btc > 0)
    # ETH stronger AND positive
    eth_wins = valid & (abs_mom_eth > abs_mom_btc) & (abs_mom_eth > 0)

    dir_btc[btc_wins] = 1.0
    dir_eth[eth_wins] = 1.0

    # Vol-target: need a df-like object with close + datetime for vol_target()
    # We rebuild minimal DataFrames aligned to the merged index
    df_btc_aligned = pd.DataFrame({
        "close": c_btc,
        "datetime": merged["datetime"].values
    })
    df_eth_aligned = pd.DataFrame({
        "close": c_eth,
        "datetime": merged["datetime"].values
    })

    tgt_btc = al.vol_target(dir_btc, df_btc_aligned,
                             target_vol=target_vol,
                             vol_win_days=30,
                             leverage_cap=leverage_cap)
    tgt_eth = al.vol_target(dir_eth, df_eth_aligned,
                             target_vol=target_vol,
                             vol_win_days=30,
                             leverage_cap=leverage_cap)

    # Return targets keyed by datetime for alignment back to per-asset dfs
    return merged["datetime"].values, tgt_btc, tgt_eth


# ---------------------------------------------------------------------------
# study_weights wrapper: target_fn(df) must return array of len(df).
# We pre-compute the cross-asset targets and align back using datetime.
# ---------------------------------------------------------------------------

def make_target_fn(tf: str, lookback_days: int, asset: str):
    """Return a target_fn(df) -> array for a given tf, lookback, asset."""

    def target_fn(df):
        dts, tgt_btc, tgt_eth = dual_momentum_targets(tf, lookback_days)
        # Map back to df's datetime index
        df_dt = pd.to_datetime(df["datetime"], utc=True).values

        # Build a lookup: datetime -> target
        tgt_map = dict(zip(dts, tgt_btc if asset == "BTC" else tgt_eth))

        target = np.array([tgt_map.get(dt, 0.0) for dt in df_dt], dtype=float)
        return target

    return target_fn


# ---------------------------------------------------------------------------
# Grid search: lookback windows
# ---------------------------------------------------------------------------

LOOKBACKS = [21, 63, 126]   # ~1m, 3m, 6m in trading days
TFS = ("1d", "12h")         # 3 lookbacks x 2 TFs x 2 assets = 12 backtests but study_weights
                              # runs both assets internally; so 3 lookbacks x 2 TFs = 6 calls

results_by_config = {}

for lb in LOOKBACKS:
    for tf in TFS:
        config_name = f"XAS09-dm-lb{lb}d-{tf}"
        print(f"\nRunning {config_name}...")

        # Precompute cross-asset targets for this tf+lb
        try:
            dts, tgt_btc_arr, tgt_eth_arr = dual_momentum_targets(tf, lb)
        except Exception as e:
            print(f"  ERROR computing targets: {e}")
            continue

        # Build per-asset target lookups (datetime -> position)
        tgt_map_btc = dict(zip(dts, tgt_btc_arr))
        tgt_map_eth = dict(zip(dts, tgt_eth_arr))

        def make_fn(asset_map):
            def fn(df):
                df_dt = pd.to_datetime(df["datetime"], utc=True).values
                return np.array([asset_map.get(dt, 0.0) for dt in df_dt], dtype=float)
            return fn

        fn_btc = make_fn(tgt_map_btc)
        fn_eth = make_fn(tgt_map_eth)

        # We need to call study_weights but with different fns per asset.
        # study_weights calls target_fn(df) for each asset, so we use a dispatch fn:
        tgt_map_by_asset = {"BTC": tgt_map_btc, "ETH": tgt_map_eth}

        def dispatch_fn(df, _maps=tgt_map_by_asset, _tf=tf, _lb=lb):
            # Detect which asset this df corresponds to by checking close range
            # Actually, we dispatch using a closure trick: call differently.
            # We can't know the asset name inside study_weights easily.
            # Workaround: check if median close > 10000 → BTC, else ETH
            med_close = np.median(df["close"].values)
            asset_key = "BTC" if med_close > 10000 else "ETH"
            amap = _maps[asset_key]
            df_dt = pd.to_datetime(df["datetime"], utc=True).values
            return np.array([amap.get(dt, 0.0) for dt in df_dt], dtype=float)

        rep = al.study_weights(config_name, dispatch_fn, tfs=(tf,))
        results_by_config[config_name] = (lb, tf, rep)
        print(al.fmt(rep))


# ---------------------------------------------------------------------------
# Pick best config (highest min_asset_holdout_sharpe)
# ---------------------------------------------------------------------------

if results_by_config:
    best_name = max(
        results_by_config,
        key=lambda k: results_by_config[k][2]["verdict"].get("best_holdout_sharpe", -99)
    )
    best_lb, best_tf, best_rep = results_by_config[best_name]
    print(f"\n\n=== BEST CONFIG: {best_name} ===")
    print(al.fmt(best_rep))
    print("JSON:", al.as_json(best_rep))
else:
    print("ERROR: no configs completed successfully")