"""XAS03 — RS Rotation BTC/ETH IDEA: Hold whichever of BTC/ETH has the stronger 90d momentum; vol-targeted; flat if both have negative momentum. The portfolio is long 1 asset at a time (or flat). IMPLEMENTATION: - Align BTC and ETH on timestamp (inner join). - Compute 90d return (close[i] / close[i - lookback] - 1) for each asset at each bar. - Winner = asset with higher momentum IF > 0; otherwise flat. - Build a combined portfolio return = winner's return at each bar. - Apply vol-targeting on the portfolio return series. - Evaluate on the combined (portfolio) return series. GRID (<=4 configs, TF 1d only -> 4 backtests within limit): lookback_days: [60, 90, 120, 180] (vol_target fixed at 20%, leverage_cap 2x) The rotation portfolio is a single return stream (not per-asset), so we build a synthetic df with close = cumulative product of the portfolio returns, then call eval_weights. This is honest: decision at bar i uses close[i], position held during bar i+1. """ 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 # =========================================================================== # Core: build aligned BTC+ETH df and compute rotation portfolio # =========================================================================== def build_rotation_df(tf: str) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]: """Merge BTC and ETH on timestamp (inner join). Return merged, btc, eth sub-dfs.""" btc = al.get("BTC", tf)[["timestamp", "datetime", "close"]].rename(columns={"close": "btc_close"}) eth = al.get("ETH", tf)[["timestamp", "close"]].rename(columns={"close": "eth_close"}) merged = pd.merge(btc, eth, on="timestamp", how="inner").reset_index(drop=True) return merged def make_rotation_target(merged: pd.DataFrame, lookback_days: int, target_vol: float = 0.20, vol_win_days: int = 30, leverage_cap: float = 2.0) -> np.ndarray: """ At each bar i, compare BTC and ETH 'lookback_days'-day momentum. Winner = asset with stronger (higher) momentum IF positive; flat if both negative. Returns the vol-targeted position in the PORTFOLIO (which itself is long BTC or ETH). We build a synthetic close = cumulative portfolio for vol-targeting. The decision (winner) is made with data up to close[i], so the position is held at bar i+1. The eval_weights shift handles this correctly. To apply vol_target over the portfolio, we compute the portfolio's own realized vol. Since we cannot run eval_weights before deciding positions, we use a simpler approach: apply vol-target scaling based on the WINNER's individual realized vol at decision time. This is still causal: vol_win_days realized vol of whichever asset we're scaling. """ btc = merged["btc_close"].values.astype(float) eth = merged["eth_close"].values.astype(float) n = len(merged) # Infer bars per day from datetime col dt_series = pd.to_datetime(merged["datetime"], utc=True) dt_diff_s = dt_series.diff().dt.total_seconds().median() bpd = max(1, round(86400 / dt_diff_s)) if dt_diff_s and dt_diff_s > 0 else 1 bpy = bpd * 365.25 lookback = max(2, lookback_days * bpd) vol_win = max(2, vol_win_days * bpd) # Simple returns for vol estimation r_btc = np.zeros(n); r_btc[1:] = btc[1:] / btc[:-1] - 1.0 r_eth = np.zeros(n); r_eth[1:] = eth[1:] / eth[:-1] - 1.0 # Realized vol (annualized) — causal, using returns up to i inclusive rv_btc = pd.Series(r_btc).rolling(vol_win, min_periods=max(2, vol_win // 2)).std().values * np.sqrt(bpy) rv_eth = pd.Series(r_eth).rolling(vol_win, min_periods=max(2, vol_win // 2)).std().values * np.sqrt(bpy) # Momentum: close[i] / close[i - lookback] - 1 (causal: known at close[i]) mom_btc = np.full(n, np.nan) mom_eth = np.full(n, np.nan) mom_btc[lookback:] = btc[lookback:] / btc[:-lookback] - 1.0 mom_eth[lookback:] = eth[lookback:] / eth[:-lookback] - 1.0 # Rotation decision + vol scaling target = np.zeros(n) for i in range(lookback, n): mb = mom_btc[i] me = mom_eth[i] # Both negative -> flat if (not np.isfinite(mb)) or (not np.isfinite(me)): target[i] = 0.0 continue if mb <= 0.0 and me <= 0.0: target[i] = 0.0 continue # Pick winner; if one is negative and other positive, pick the positive one if mb >= me: # Go long BTC vol = rv_btc[i] direction = 1.0 else: # Go long ETH vol = rv_eth[i] direction = 1.0 # Vol target scaling if np.isfinite(vol) and vol > 0: scale = min(target_vol / vol, leverage_cap) else: scale = 0.0 target[i] = direction * scale return target def build_portfolio_df(merged: pd.DataFrame, lookback_days: int, target_vol: float = 0.20, vol_win_days: int = 30, leverage_cap: float = 2.0): """ Build the rotation portfolio: - At each bar i, compute the target (from make_rotation_target). - The target represents the fraction of equity to allocate to the winning asset. - The actual P&L at bar i+1 is: target[i] * r_winner[i+1] - We build a synthetic close series = cumulative equity of the portfolio. - Then eval_weights on this synthetic df reproduces that P&L correctly. BUT there's a subtlety: target[i] can refer to BTC OR ETH depending on the rotation. The synthetic "close" trick only works if we build the actual portfolio returns directly. BETTER APPROACH: compute the portfolio net returns directly, then build a synthetic df with cumulative returns as the close. eval_weights on a buy-and-hold of this df (target=1) will then give us exactly those returns (since pos=1 * r_synthetic = portfolio return). Actually, the cleanest honest approach: 1. Compute rotation signal at i (uses data <= i). 2. Portfolio gross return at bar i+1 = signal[i] * r_winner[i+1]. 3. Fee at turnover = |signal[i] - signal[i-1]| * fee_side. We do this directly and compute metrics without using eval_weights' shift (we handle the shift manually here by computing returns one step ahead). """ btc = merged["btc_close"].values.astype(float) eth = merged["eth_close"].values.astype(float) n = len(merged) dt_series = pd.to_datetime(merged["datetime"], utc=True) dt_diff_s = dt_series.diff().dt.total_seconds().median() bpd = max(1, round(86400 / dt_diff_s)) if dt_diff_s and dt_diff_s > 0 else 1 bpy = bpd * 365.25 lookback = max(2, lookback_days * bpd) vol_win = max(2, vol_win_days * bpd) r_btc = np.zeros(n); r_btc[1:] = btc[1:] / btc[:-1] - 1.0 r_eth = np.zeros(n); r_eth[1:] = eth[1:] / eth[:-1] - 1.0 rv_btc = pd.Series(r_btc).rolling(vol_win, min_periods=max(2, vol_win // 2)).std().values * np.sqrt(bpy) rv_eth = pd.Series(r_eth).rolling(vol_win, min_periods=max(2, vol_win // 2)).std().values * np.sqrt(bpy) mom_btc = np.full(n, np.nan) mom_eth = np.full(n, np.nan) mom_btc[lookback:] = btc[lookback:] / btc[:-lookback] - 1.0 mom_eth[lookback:] = eth[lookback:] / eth[:-lookback] - 1.0 # Signal at bar i (decided with data <= close[i]) # 0 = flat, 1 = long BTC, 2 = long ETH signal_dir = np.zeros(n, dtype=int) # 0=flat, 1=BTC, 2=ETH signal_size = np.zeros(n) # vol-targeted position size for i in range(lookback, n): mb = mom_btc[i] me = mom_eth[i] if (not np.isfinite(mb)) or (not np.isfinite(me)): continue if mb <= 0.0 and me <= 0.0: continue if mb >= me: signal_dir[i] = 1 # BTC vol = rv_btc[i] else: signal_dir[i] = 2 # ETH vol = rv_eth[i] if np.isfinite(vol) and vol > 0: scale = min(target_vol / vol, leverage_cap) else: scale = 0.0 signal_size[i] = scale # Portfolio return at bar t = signal_size[t-1] * r_winner[t] # where winner is determined by signal_dir[t-1] port_gross = np.zeros(n) for t in range(1, n): if signal_dir[t-1] == 1: port_gross[t] = signal_size[t-1] * r_btc[t] elif signal_dir[t-1] == 2: port_gross[t] = signal_size[t-1] * r_eth[t] # else 0 # Fee on turnover: size changes + asset switches turn = np.zeros(n) prev_size = 0.0 prev_dir = 0 for t in range(1, n): cur_dir = signal_dir[t-1] cur_size = signal_size[t-1] if cur_dir != prev_dir: # Full switch: close old + open new turn[t] = prev_size + cur_size else: turn[t] = abs(cur_size - prev_size) prev_size = cur_size prev_dir = cur_dir port_net = port_gross - al.FEE_SIDE * turn # Build synthetic df with close = cumulative equity idx = dt_series return port_net, turn, idx, bpy def eval_rotation(merged: pd.DataFrame, lookback_days: int, target_vol: float = 0.20, vol_win_days: int = 30, leverage_cap: float = 2.0, fee_side: float = al.FEE_SIDE) -> dict: """Evaluate the rotation portfolio, re-scaling fee by ratio to default fee.""" btc = merged["btc_close"].values.astype(float) eth = merged["eth_close"].values.astype(float) n = len(merged) dt_series = pd.to_datetime(merged["datetime"], utc=True) dt_diff_s = dt_series.diff().dt.total_seconds().median() bpd = max(1, round(86400 / dt_diff_s)) if dt_diff_s and dt_diff_s > 0 else 1 bpy = bpd * 365.25 lookback = max(2, lookback_days * bpd) vol_win = max(2, vol_win_days * bpd) r_btc = np.zeros(n); r_btc[1:] = btc[1:] / btc[:-1] - 1.0 r_eth = np.zeros(n); r_eth[1:] = eth[1:] / eth[:-1] - 1.0 rv_btc = pd.Series(r_btc).rolling(vol_win, min_periods=max(2, vol_win // 2)).std().values * np.sqrt(bpy) rv_eth = pd.Series(r_eth).rolling(vol_win, min_periods=max(2, vol_win // 2)).std().values * np.sqrt(bpy) mom_btc = np.full(n, np.nan) mom_eth = np.full(n, np.nan) mom_btc[lookback:] = btc[lookback:] / btc[:-lookback] - 1.0 mom_eth[lookback:] = eth[lookback:] / eth[:-lookback] - 1.0 signal_dir = np.zeros(n, dtype=int) signal_size = np.zeros(n) for i in range(lookback, n): mb = mom_btc[i] me = mom_eth[i] if (not np.isfinite(mb)) or (not np.isfinite(me)): continue if mb <= 0.0 and me <= 0.0: continue if mb >= me: signal_dir[i] = 1 vol = rv_btc[i] else: signal_dir[i] = 2 vol = rv_eth[i] if np.isfinite(vol) and vol > 0: scale = min(target_vol / vol, leverage_cap) else: scale = 0.0 signal_size[i] = scale port_gross = np.zeros(n) for t in range(1, n): if signal_dir[t-1] == 1: port_gross[t] = signal_size[t-1] * r_btc[t] elif signal_dir[t-1] == 2: port_gross[t] = signal_size[t-1] * r_eth[t] turn = np.zeros(n) prev_size = 0.0 prev_dir = 0 for t in range(1, n): cur_dir = signal_dir[t-1] cur_size = signal_size[t-1] if cur_dir != prev_dir: turn[t] = prev_size + cur_size else: turn[t] = abs(cur_size - prev_size) prev_size = cur_size prev_dir = cur_dir port_net = port_gross - fee_side * turn idx = dt_series full = al._metrics_from_net(port_net, pd.DatetimeIndex(idx)) hmask = idx >= al.HOLDOUT hold = al._metrics_from_net(port_net[hmask], pd.DatetimeIndex(idx[hmask])) if hmask.sum() > 3 \ else dict(sharpe=0.0, n=0) # Yearly s = pd.Series(np.nan_to_num(port_net), index=pd.DatetimeIndex(idx)) yearly = {} for y, g in s.groupby(s.index.year): eq = np.cumprod(1 + g.values) pk = np.maximum.accumulate(eq) yearly[int(y)] = dict(ret=round(float(eq[-1] - 1), 4), dd=round(float(np.max((pk - eq) / pk)), 4)) tpy = float(turn.sum() / (len(turn) / bpy)) if len(turn) > 0 else 0.0 tim = float(np.mean(signal_dir > 0)) return dict(full=full, holdout=hold, yearly=yearly, time_in_market=round(tim, 3), turnover_per_year=round(tpy, 1), port_net=port_net, idx=idx) # =========================================================================== # Grid search: 4 lookback configs on 1d TF # =========================================================================== TFS = ("1d",) GRID = [ {"lookback_days": 60}, {"lookback_days": 90}, {"lookback_days": 120}, {"lookback_days": 180}, ] print("=== XAS03: RS Rotation BTC/ETH ===") print(f"Grid: {len(GRID)} lookbacks x {len(TFS)} TFs = {len(GRID)*len(TFS)} backtests") print() best_rep = None best_score = -999.0 best_label = "" for tf in TFS: merged = build_rotation_df(tf) print(f"TF={tf}: {len(merged)} aligned bars, " f"{merged['datetime'].iloc[0]} -> {merged['datetime'].iloc[-1]}") for params in GRID: lb = params["lookback_days"] name = f"XAS03-lb{lb}-{tf}" print(f"\n--- {name} ---") base = eval_rotation(merged, lb) fee_sweep = {} for f in al.FEE_SWEEP: sh = eval_rotation(merged, lb, fee_side=f)["full"]["sharpe"] fee_sweep[f"{2*f*100:.2f}%RT"] = sh fee_ok = fee_sweep.get("0.20%RT", -9) > 0 full = base["full"] hold = base["holdout"] yearly = base["yearly"] print(f" full Sh={full['sharpe']:+.3f} DD={full['maxdd']*100:.1f}% ret={full['ret']*100:+.0f}%") print(f" hold Sh={hold.get('sharpe',0):+.3f} ret={hold.get('ret',0)*100:+.0f}%") print(f" time_in_market={base['time_in_market']:.2f} turnover/yr={base['turnover_per_year']:.1f}") print(f" fee sweep: {fee_sweep}") yr_str = " ".join(f"{y}:{v['ret']*100:+.0f}%" for y, v in sorted(yearly.items())) print(f" yearly: {yr_str}") # The rotation portfolio is evaluated as a single entity. # For compatibility with al.fmt, we replicate it as both BTC and ETH entries # since it IS the portfolio of those two assets. per_asset_result = dict(full=full, holdout=hold, tim=base["time_in_market"], turnover=base["turnover_per_year"], fee_sweep=fee_sweep, yearly=yearly) cells = [dict( tf=tf, per_asset={"BTC": per_asset_result, "ETH": per_asset_result}, 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, )] rep = dict(name=name, kind="weights", cells=cells, verdict=al._verdict(cells)) score = hold.get("sharpe", 0.0) if score > best_score: best_score = score best_label = name best_rep = rep print() print("=" * 60) print("BEST CONFIG:", best_label) print(al.fmt(best_rep)) print() print("JSON:", al.as_json(best_rep))