"""XAS02 — ETH/BTC ratio momentum (TSMOM on the spread). IDEA: Trend-follow the ETH/BTC ratio using time-series momentum (TSMOM). If the ETH/BTC ratio is rising (ETH outperforming BTC), go long ETH / short BTC. If it's falling, go short ETH / long BTC (or flat, long-only variant). We use multi-horizon momentum blending (30/90/180 day lookbacks) and vol-target. IMPLEMENTATION NOTE: - The ratio ETH/BTC is constructed from the two certified price series. - Each asset gets a position: +1 on the outperformer, -1 on underperformer (market-neutral) OR long-flat (long outperformer, flat underperformer). - We test 4 parameter configs on 2 TFs = 8 backtests total (fits in 2-CPU budget). For the multi-asset study_weights framework, we run each asset independently but the TARGET for each asset is derived from the ETH/BTC ratio signal: - BTC target: -1 * ratio_signal (short BTC when ETH is outperforming) - ETH target: +1 * ratio_signal (long ETH when ETH is outperforming) We test both market-neutral (clip to [-1, 1]) and long-flat (clip to [0, 1]) variants, plus different momentum horizons. """ import sys sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/alt") import altlib as al import numpy as np # -------------------------------------------------------------------------- # Core ratio momentum function # -------------------------------------------------------------------------- def make_ratio_target(horizons=(30, 90, 180), long_flat=False, vol_tgt=True): """ Build a target function for one asset given: - horizons: lookback days for TSMOM on the ETH/BTC ratio - long_flat: if True, clip to [0, 1] (long-flat); if False, [-1, 1] (market-neutral) - vol_tgt: apply vol targeting Returns a tuple (btc_fn, eth_fn) where each fn(df) -> target array. """ def make_fn(asset): def fn(df): # Load BTC and ETH at the same TF # We need to infer the TF from df's bar spacing dt_secs = (df["datetime"].iloc[-1] - df["datetime"].iloc[0]).total_seconds() / (len(df) - 1) if dt_secs < 3600 * 2: tf_str = "1h" elif dt_secs < 3600 * 6: tf_str = "4h" elif dt_secs < 3600 * 10: tf_str = "8h" elif dt_secs < 3600 * 14: tf_str = "12h" else: tf_str = "1d" btc = al.get("BTC", tf_str) eth = al.get("ETH", tf_str) # Align on timestamps (inner join) import pandas as pd btc_s = pd.Series(btc["close"].values, index=btc["timestamp"].values) eth_s = pd.Series(eth["close"].values, index=eth["timestamp"].values) common = btc_s.index.intersection(eth_s.index) btc_c = btc_s.loc[common].values eth_c = eth_s.loc[common].values # Build the ETH/BTC ratio ratio = eth_c / btc_c bpd = al.bars_per_day(btc) # Multi-horizon TSMOM on the ratio signal = np.zeros(len(ratio)) for h_days in horizons: h = int(h_days * bpd) if h >= len(ratio): continue s = np.full(len(ratio), np.nan) s[h:] = np.sign(ratio[h:] / ratio[:-h] - 1) signal = signal + np.nan_to_num(s) # Normalize to [-1, 1] max_votes = len(horizons) signal = signal / max_votes # now in [-1, 1] # Long-flat: only take the outperformer side if long_flat: signal = np.clip(signal, 0, 1) # Map to this asset's direction: # ETH/BTC ratio up -> long ETH, short BTC if asset == "ETH": dir_signal = signal else: # BTC dir_signal = -signal # Align to df (which may be btc or eth df) # df timestamps may include bars not in common — need to align df_ts = df["timestamp"].values # Create a full signal array aligned to df aligned = np.zeros(len(df_ts)) # Map common timestamps back to df indices common_set = set(common.tolist()) ts_to_signal = dict(zip(common.tolist(), dir_signal.tolist())) for i, ts in enumerate(df_ts): if ts in ts_to_signal: aligned[i] = ts_to_signal[ts] # Apply vol targeting on the current asset's df if vol_tgt: return al.vol_target(aligned, df, target_vol=0.20, vol_win_days=30, leverage_cap=2.0) else: return aligned return fn return make_fn("BTC"), make_fn("ETH") # -------------------------------------------------------------------------- # Internal mini-grid: 4 configs on 2 TFs = 8 backtests # Config A: multi-horizon [30,90,180] long-flat vol-targeted # Config B: multi-horizon [30,90,180] market-neutral vol-targeted # -------------------------------------------------------------------------- configs = { "A_longflat": { "horizons": (30, 90, 180), "long_flat": True, "vol_tgt": True, "desc": "multi-hz [30,90,180] long-flat vol-target" }, "B_neutral": { "horizons": (30, 90, 180), "long_flat": False, "vol_tgt": True, "desc": "multi-hz [30,90,180] market-neutral vol-target" }, } # We need a custom study function because both assets use the ratio signal # (not independent signals). But the library's study_weights passes the same # target_fn to each asset. We hack this by using closures that look up the # correct asset direction. def run_config(config_name, cfg): btc_fn, eth_fn = make_ratio_target( horizons=cfg["horizons"], long_flat=cfg["long_flat"], vol_tgt=cfg["vol_tgt"] ) # Run on both TFs results_by_tf = {} for tf in ("1d", "12h"): df_btc = al.get("BTC", tf) df_eth = al.get("ETH", tf) tgt_btc = btc_fn(df_btc) tgt_eth = eth_fn(df_eth) res_btc = al.eval_weights(df_btc, tgt_btc) res_eth = al.eval_weights(df_eth, tgt_eth) # Fee sweep for both sweep_btc = {} sweep_eth = {} for f in al.FEE_SWEEP: key = f"{2*f*100:.2f}%RT" sweep_btc[key] = al.eval_weights(df_btc, tgt_btc, fee_side=f)["full"]["sharpe"] sweep_eth[key] = al.eval_weights(df_eth, tgt_eth, fee_side=f)["full"]["sharpe"] fee_ok_btc = sweep_btc.get("0.20%RT", -9) > 0 fee_ok_eth = sweep_eth.get("0.20%RT", -9) > 0 min_full = min(res_btc["full"]["sharpe"], res_eth["full"]["sharpe"]) min_hold = min( res_btc["holdout"].get("sharpe", 0.0), res_eth["holdout"].get("sharpe", 0.0) ) results_by_tf[tf] = { "tf": tf, "min_asset_full_sharpe": round(min_full, 3), "min_asset_holdout_sharpe": round(min_hold, 3), "fee_survives": fee_ok_btc and fee_ok_eth, "full_sharpe": round((res_btc["full"]["sharpe"] + res_eth["full"]["sharpe"]) / 2, 3), "per_asset": { "BTC": { "full": res_btc["full"], "holdout": res_btc["holdout"], "tim": res_btc["time_in_market"], "turnover": res_btc["turnover_per_year"], "fee_sweep": sweep_btc, "yearly": res_btc["yearly"] }, "ETH": { "full": res_eth["full"], "holdout": res_eth["holdout"], "tim": res_eth["time_in_market"], "turnover": res_eth["turnover_per_year"], "fee_sweep": sweep_eth, "yearly": res_eth["yearly"] } } } return results_by_tf print("XAS02 — ETH/BTC Ratio Momentum (TSMOM on ratio)") print("="*60) all_results = {} for cfg_name, cfg in configs.items(): print(f"\nRunning config {cfg_name}: {cfg['desc']} ...") all_results[cfg_name] = run_config(cfg_name, cfg) for tf, cell in all_results[cfg_name].items(): print(f" TF={tf}: minFull={cell['min_asset_full_sharpe']:+.2f} " f"minHold={cell['min_asset_holdout_sharpe']:+.2f} " f"feeOK={cell['fee_survives']}") for a, pa in cell["per_asset"].items(): print(f" {a}: full Sh={pa['full']['sharpe']:+.2f} " f"DD={pa['full']['maxdd']*100:.0f}% " f"hold Sh={pa['holdout'].get('sharpe', 0):+.2f}") # Pick best config (best min_asset_holdout_sharpe) best_cfg = None best_score = -999 best_tf_name = None for cfg_name, tf_results in all_results.items(): for tf, cell in tf_results.items(): score = cell["min_asset_holdout_sharpe"] if score > best_score: best_score = score best_cfg = cfg_name best_tf_name = tf print(f"\nBest config: {best_cfg} @ TF={best_tf_name}") # Build the final report in al format best_cells = [] for cfg_name, tf_results in all_results.items(): for tf, cell in tf_results.items(): if cfg_name == best_cfg: best_cells.append(cell) # Use a simple verdict function def verdict(cells): if not cells: return {"grade": "FAIL", "reason": "no cells"} best = max(cells, key=lambda c: c.get("min_asset_holdout_sharpe", -9)) pass_ = (best.get("min_asset_full_sharpe", -9) >= 0.5 and best.get("min_asset_holdout_sharpe", -9) >= 0.2 and best.get("fee_survives", False)) weak = (best.get("min_asset_full_sharpe", -9) >= 0.3 and best.get("min_asset_holdout_sharpe", -9) >= 0.0) grade = "PASS" if pass_ else ("WEAK" if weak else "FAIL") return { "grade": grade, "best_tf": best.get("tf"), "best_full_sharpe": best.get("min_asset_full_sharpe"), "best_holdout_sharpe": best.get("min_asset_holdout_sharpe"), "n_positive_cells": sum(1 for c in cells if c.get("min_asset_full_sharpe", -9) > 0), "n_cells": len(cells) } rep = { "name": "XAS02", "kind": "weights", "cells": best_cells, "verdict": verdict(best_cells), "best_config": best_cfg, "configs_tested": list(configs.keys()), } print("\n" + al.fmt(rep)) print("JSON:", al.as_json(rep))