Adriano Dal Pastro
5ac4e16af8
Ondata di ricerca onesta a largo spettro su BTC/ETH+DVOL certificati: 104 ipotesi distinte (11 famiglie), un agente-finder per ipotesi, verifica avversariale a 3 scettici sui promettenti, sintesi (153 agenti totali). Esito: NIENTE di nuovo regge -> conferma del soffitto strutturale ~1.3 BTC/ETH-direzionale; lo stack TP01+XS01+VRP01 resta imbattuto. - altlib.py: harness condiviso vettoriale leak-free (eval_weights/study_weights, fee-sweep, both-asset + hold-out 2025+). Riproduce i numeri canonici di TP01. - MARGINAL SCORER (study_marginal/marginal_vs_tp01): Sharpe INCREMENTALE vs baseline TP01 (corr, blend uplift OOS, alpha residua) + jackknife OOS (clean-year + drop-best-month). earns_slot = abs!=FAIL & ADDS & robust_oos. Smaschera gli overlay su TSMOM con PASS assoluti fasulli (CMB04, VOL11, ...) e il falso positivo KAMA (ADDS ma muore al jackknife). - runs/*.py (104) script riproducibili per ipotesi; wf_altstrat.js workflow. - Verdetto: 0 candidati deployabili; 2 LEAD fragili (VOL08, STA05_LS) da forward-monitor. - test_marginal_scorer.py blocca baseline + invarianti. Suite: 32 verde. Diario: docs/diary/2026-06-20-alt-strategies-100agent-sweep.md Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
393 lines
15 KiB
Python
393 lines
15 KiB
Python
"""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))
|