Adriano
58fc10de77
- trackH volume_vol: no uncorrelated additive edge; profitable signals are trend-in-disguise (corr 0.6-0.75); MR/declining-volume fade dead even at fee 0; OBV-up filter is a defensive DD overlay only (13.3->10.1% DD but -CAGR), not new alpha - trackI momentum/reversal: no formulation beats 1-3-6m sign-blend OOS on both assets; z-score continuous momentum = same edge (corr 0.96), lower DD 8.4% but lower CAGR; long-horizon reversal not bankable (negative/flat standalone). ~1.3 Sharpe ceiling holds. - TP01 (12h sign-blend) remains the deployable winner
603 lines
26 KiB
Python
603 lines
26 KiB
Python
"""TRACK H — VOLUME, RANGE & VOLATILITY-REGIME signals on CLEAN BTC/ETH (Deribit mainnet).
|
||
|
||
The single question: net of realistic Deribit fees, OOS-robust on BOTH BTC & ETH, on >=12h
|
||
timeframes (the only honest regime — sub-12h is fees + HF-noise overfit + the open-label
|
||
look-ahead trap), is there ANY volume / range / volatility-regime signal that is
|
||
|
||
(a) net-positive OOS on both assets standalone, AND
|
||
(b) uncorrelated (|corr| < ~0.3) to the deployed winner TP01, AND/OR
|
||
(c) usable as a REGIME FILTER that lifts TP01's Sharpe above ~1.32 or cuts its DD?
|
||
|
||
HONESTY / NO LOOK-AHEAD:
|
||
* Everything runs on the SAME causal per-bar engine used by TP01 (net_returns): we build a
|
||
continuous TARGET position decided with data <= close[i], then HOLD it during bar i+1
|
||
(pos_held[t] = target[t-1]). Gross = pos_held * simple_return[t]; fee charged on |Δpos|.
|
||
This is identical in spirit to the harness `backtest_signals` (decide<=close[i], fill at
|
||
close[i]); we cross-check two discrete signals through `backtest_signals` too.
|
||
* Volume / range / vol features for bar i use ONLY bars <= i (rolling, prior-window, shift).
|
||
* 12h / 1d frames are resampled from the certified 1h feed via resample_tf (label='left',
|
||
closed='left') and consumed index-based with the +1 bar hold -> the open-label is never
|
||
leaked (verified in trackD_lookahead_audit.py: Sharpe is label-invariant under this hold).
|
||
|
||
Run:
|
||
uv run python scripts/research/trackH_volume_vol.py # full (12h + 1d)
|
||
uv run python scripts/research/trackH_volume_vol.py --quick # 12h only, fewer grids
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||
|
||
from src.backtest.harness import load, backtest_signals
|
||
from src.strategies.trend_portfolio import TrendPortfolio, CANONICAL, resample_tf
|
||
|
||
ASSETS = ["BTC", "ETH"]
|
||
FEE_SIDE = 0.0005 # 0.05%/side = 0.10% RT (Deribit taker)
|
||
OOS_FRAC = 0.65
|
||
TF_BPD = {"12h": 2, "1d": 1}
|
||
|
||
|
||
# ===========================================================================
|
||
# Causal feature helpers (all use data <= i)
|
||
# ===========================================================================
|
||
def simple_returns(c: np.ndarray) -> np.ndarray:
|
||
r = np.zeros(len(c))
|
||
r[1:] = c[1:] / c[:-1] - 1.0
|
||
return r
|
||
|
||
|
||
def realized_vol(r: np.ndarray, win: int, bpy: float) -> np.ndarray:
|
||
return pd.Series(r).rolling(win, min_periods=max(2, win // 2)).std().values * np.sqrt(bpy)
|
||
|
||
|
||
def roll_max_prior(x: np.ndarray, win: int) -> np.ndarray:
|
||
"""Max over the PRIOR `win` bars (excludes current bar i)."""
|
||
return pd.Series(x).shift(1).rolling(win, min_periods=win).max().values
|
||
|
||
|
||
def roll_min_prior(x: np.ndarray, win: int) -> np.ndarray:
|
||
return pd.Series(x).shift(1).rolling(win, min_periods=win).min().values
|
||
|
||
|
||
def roll_mean_prior(x: np.ndarray, win: int) -> np.ndarray:
|
||
return pd.Series(x).shift(1).rolling(win, min_periods=win).mean().values
|
||
|
||
|
||
def vol_zscore(vol: np.ndarray, win: int) -> np.ndarray:
|
||
"""z-score of current volume vs PRIOR `win` bars (uses <= i)."""
|
||
s = pd.Series(vol)
|
||
m = s.shift(1).rolling(win, min_periods=win).mean()
|
||
sd = s.shift(1).rolling(win, min_periods=win).std()
|
||
return ((s - m) / sd).values
|
||
|
||
|
||
def atr(df: pd.DataFrame, period: int) -> np.ndarray:
|
||
h, l, c = df["high"].values, df["low"].values, df["close"].values
|
||
pc = np.roll(c, 1)
|
||
pc[0] = c[0]
|
||
tr = np.maximum(h - l, np.maximum(np.abs(h - pc), np.abs(l - pc)))
|
||
return pd.Series(tr).ewm(alpha=1.0 / period, adjust=False).mean().values
|
||
|
||
|
||
# ===========================================================================
|
||
# Per-bar net-returns engine (causal, fee on turnover) — identical to TP01.net_returns
|
||
# ===========================================================================
|
||
def net_from_target(target: np.ndarray, r: np.ndarray, fee_side: float):
|
||
"""target[i] decided with data <= close[i] -> HELD during bar i+1."""
|
||
target = np.nan_to_num(target, nan=0.0)
|
||
pos = np.zeros(len(target))
|
||
pos[1:] = target[:-1]
|
||
gross = pos * r
|
||
turn = np.abs(np.diff(pos, prepend=0.0))
|
||
net = gross - fee_side * turn
|
||
net[0] = 0.0
|
||
net = np.clip(net, -0.99, None)
|
||
return net, pos, turn
|
||
|
||
|
||
def metrics(net: np.ndarray, idx: pd.DatetimeIndex, turn: np.ndarray, bpy: float) -> dict:
|
||
rr = net[np.isfinite(net)]
|
||
sharpe = float(np.mean(rr) / np.std(rr) * np.sqrt(bpy)) if np.std(rr) > 0 else 0.0
|
||
equity = np.cumprod(1.0 + np.clip(net, -0.99, None))
|
||
peak = np.maximum.accumulate(equity)
|
||
dd = float(np.max((peak - equity) / peak)) if len(equity) else 0.0
|
||
span_days = (idx[-1] - idx[0]).total_seconds() / 86400
|
||
years = span_days / 365.25 if span_days > 0 else 1.0
|
||
total = equity[-1] / equity[0] if len(equity) else 1.0
|
||
cagr = total ** (1 / years) - 1 if years > 0 and total > 0 else -1.0
|
||
ann_turn = float(np.sum(turn)) / years if years > 0 else 0.0
|
||
return dict(sharpe=sharpe, max_dd=dd, cagr=cagr, total=total - 1,
|
||
ann_turnover=ann_turn, equity=equity, years=years)
|
||
|
||
|
||
def per_year(net: np.ndarray, idx: pd.DatetimeIndex) -> dict:
|
||
eq = pd.Series(np.cumprod(1.0 + np.clip(net, -0.99, None)), index=idx)
|
||
out = {}
|
||
for y, g in eq.groupby(eq.index.year):
|
||
if len(g) > 1 and g.iloc[0] > 0:
|
||
out[int(y)] = float(g.iloc[-1] / g.iloc[0] - 1)
|
||
return out
|
||
|
||
|
||
# ===========================================================================
|
||
# SIGNALS — each returns a continuous TARGET array (frac of equity, +/-), causal.
|
||
# ===========================================================================
|
||
def sig_vt_long(df, bpd, target_vol=0.20, vol_win_days=30, lev=2.0, **_):
|
||
"""Volatility-managed LONG: always long, sized to a vol target (no trend at all).
|
||
Tests Moreira-Muir 'volatility-managed' alpha vs plain buy-and-hold."""
|
||
c = df["close"].values.astype(float)
|
||
r = simple_returns(c)
|
||
bpy = bpd * 365.25
|
||
vol = realized_vol(r, vol_win_days * bpd, bpy)
|
||
tgt = np.where((vol > 0) & np.isfinite(vol), target_vol / vol, 0.0)
|
||
return np.clip(tgt, 0, lev)
|
||
|
||
|
||
def sig_vol_breakout(df, bpd, don=20, zwin=20, zk=1.0, long_short=False, **_):
|
||
"""Volume-confirmed Donchian breakout (continuation). Long when close > prior-`don`-bar high
|
||
AND volume z-score > zk; stay long until close < prior-`don`-bar low (then flat/short)."""
|
||
c = df["close"].values.astype(float)
|
||
h = df["high"].values.astype(float)
|
||
l = df["low"].values.astype(float)
|
||
vol = df["volume"].values.astype(float)
|
||
hi = roll_max_prior(h, don)
|
||
lo = roll_min_prior(l, don)
|
||
z = vol_zscore(vol, zwin)
|
||
up = (c > hi) & (z > zk)
|
||
dn = (c < lo) & (z > zk)
|
||
state = np.zeros(len(c))
|
||
s = 0.0
|
||
for i in range(len(c)):
|
||
if up[i]:
|
||
s = 1.0
|
||
elif dn[i]:
|
||
s = -1.0 if long_short else 0.0
|
||
elif s == 1.0 and c[i] < lo[i]: # trailing exit for longs
|
||
s = -1.0 if long_short else 0.0
|
||
elif s == -1.0 and c[i] > hi[i]:
|
||
s = 1.0
|
||
state[i] = s
|
||
return state
|
||
|
||
|
||
def sig_obv_trend(df, bpd, ma=30, long_short=False, **_):
|
||
"""OBV trend: OBV = cumsum(sign(ret)*volume); long when OBV > its EMA(ma), else flat/short."""
|
||
c = df["close"].values.astype(float)
|
||
vol = df["volume"].values.astype(float)
|
||
r = simple_returns(c)
|
||
obv = np.cumsum(np.sign(r) * vol)
|
||
ema = pd.Series(obv).ewm(span=ma, adjust=False).mean().values
|
||
d = np.where(obv > ema, 1.0, (-1.0 if long_short else 0.0))
|
||
return d
|
||
|
||
|
||
def sig_vw_momentum(df, bpd, mom_win=30, vol_win_days=30, target_vol=0.20, lev=2.0,
|
||
long_only=True, **_):
|
||
"""Volume-weighted momentum: sign of volume-weighted mean return over `mom_win` bars,
|
||
vol-targeted. Compare to plain TSMOM (does weighting by volume add anything?)."""
|
||
c = df["close"].values.astype(float)
|
||
vol = df["volume"].values.astype(float)
|
||
r = simple_returns(c)
|
||
rw = r * vol
|
||
num = pd.Series(rw).rolling(mom_win, min_periods=mom_win).sum().values
|
||
den = pd.Series(vol).rolling(mom_win, min_periods=mom_win).sum().values
|
||
vwret = np.where(den > 0, num / den, 0.0)
|
||
direction = np.sign(vwret)
|
||
if long_only:
|
||
direction = np.clip(direction, 0, None)
|
||
bpy = bpd * 365.25
|
||
rv = realized_vol(r, vol_win_days * bpd, bpy)
|
||
scal = np.where((rv > 0) & np.isfinite(rv), target_vol / rv, 0.0)
|
||
return np.clip(direction * scal, -lev, lev)
|
||
|
||
|
||
def sig_range_expansion(df, bpd, rng_win=20, k=1.5, hold=5, long_short=False, **_):
|
||
"""Range-expansion breakout: when today's range > k * avg(prior `rng_win` ranges) and the
|
||
bar closed in the upper/lower half, go with the close direction; hold `hold` bars."""
|
||
c = df["close"].values.astype(float)
|
||
h = df["high"].values.astype(float)
|
||
l = df["low"].values.astype(float)
|
||
rng = h - l
|
||
avg = roll_mean_prior(rng, rng_win)
|
||
expand = rng > k * avg
|
||
pos_in_bar = np.where(rng > 0, (c - l) / rng, 0.5)
|
||
long_trig = expand & (pos_in_bar > 0.6)
|
||
short_trig = expand & (pos_in_bar < 0.4)
|
||
state = np.zeros(len(c))
|
||
hold_left = 0
|
||
cur = 0.0
|
||
for i in range(len(c)):
|
||
if hold_left > 0:
|
||
hold_left -= 1
|
||
else:
|
||
cur = 0.0
|
||
if long_trig[i]:
|
||
cur = 1.0
|
||
hold_left = hold
|
||
elif short_trig[i] and long_short:
|
||
cur = -1.0
|
||
hold_left = hold
|
||
state[i] = cur
|
||
return state
|
||
|
||
|
||
def sig_nr_breakout(df, bpd, nr=7, hold=5, long_short=False, **_):
|
||
"""NR-N breakout (daily-style): when the current bar's range is the narrowest of the last
|
||
`nr` bars, take the breakout of the prior bar's high/low on the next bars; hold `hold`."""
|
||
c = df["close"].values.astype(float)
|
||
h = df["high"].values.astype(float)
|
||
l = df["low"].values.astype(float)
|
||
rng = h - l
|
||
is_nr = pd.Series(rng).rolling(nr, min_periods=nr).apply(
|
||
lambda w: 1.0 if w[-1] == np.min(w) else 0.0, raw=True).values
|
||
state = np.zeros(len(c))
|
||
cur = 0.0
|
||
hold_left = 0
|
||
armed = False
|
||
arm_hi = arm_lo = np.nan
|
||
for i in range(len(c)):
|
||
if hold_left > 0:
|
||
hold_left -= 1
|
||
else:
|
||
cur = 0.0
|
||
if armed:
|
||
if c[i] > arm_hi:
|
||
cur = 1.0
|
||
hold_left = hold
|
||
armed = False
|
||
elif c[i] < arm_lo and long_short:
|
||
cur = -1.0
|
||
hold_left = hold
|
||
armed = False
|
||
if is_nr[i] == 1.0:
|
||
armed = True
|
||
arm_hi = h[i]
|
||
arm_lo = l[i]
|
||
state[i] = cur
|
||
return state
|
||
|
||
|
||
def sig_decl_vol_reversal(df, bpd, mom_win=10, vwin=10, **_):
|
||
"""Declining-volume reversal (fade): after an up-move on DECLINING volume, fade (short);
|
||
after a down-move on declining volume, go long. Pure contrarian, vol-confirmed exhaustion."""
|
||
c = df["close"].values.astype(float)
|
||
vol = df["volume"].values.astype(float)
|
||
ret = pd.Series(c).pct_change(mom_win).values
|
||
vtrend = vol - roll_mean_prior(vol, vwin)
|
||
declining = vtrend < 0
|
||
state = np.zeros(len(c))
|
||
state[(ret > 0) & declining] = -1.0
|
||
state[(ret < 0) & declining] = 1.0
|
||
return state
|
||
|
||
|
||
SIGNALS = {
|
||
"VT-long": (sig_vt_long, dict(target_vol=0.20, vol_win_days=30, lev=2.0)),
|
||
"VolBreakout": (sig_vol_breakout, dict(don=20, zwin=20, zk=1.0)),
|
||
"OBV-trend": (sig_obv_trend, dict(ma=30)),
|
||
"VW-mom": (sig_vw_momentum, dict(mom_win=30, vol_win_days=30)),
|
||
"RangeExpand": (sig_range_expansion, dict(rng_win=20, k=1.5, hold=5)),
|
||
"NR7-break": (sig_nr_breakout, dict(nr=7, hold=5)),
|
||
"DeclVolRev": (sig_decl_vol_reversal, dict(mom_win=10, vwin=10)),
|
||
}
|
||
|
||
|
||
# ===========================================================================
|
||
# Evaluation
|
||
# ===========================================================================
|
||
def eval_signal(fn, params, tf, asset, fee_side=FEE_SIDE):
|
||
df = resample_tf(load(asset, "1h"), tf)
|
||
bpd = TF_BPD[tf]
|
||
bpy = bpd * 365.25
|
||
c = df["close"].values.astype(float)
|
||
r = simple_returns(c)
|
||
idx = pd.to_datetime(df["datetime"].values)
|
||
tgt = fn(df, bpd, **params)
|
||
net, pos, turn = net_from_target(tgt, r, fee_side)
|
||
m = metrics(net, idx, turn, bpy)
|
||
# OOS split
|
||
cut = int(len(net) * OOS_FRAC)
|
||
mi = metrics(net[:cut], idx[:cut], turn[:cut], bpy)
|
||
mo = metrics(net[cut:], idx[cut:], turn[cut:], bpy)
|
||
return dict(net=net, idx=idx, full=m, is_=mi, oos=mo, py=per_year(net, idx))
|
||
|
||
|
||
def tp01_net(asset, tf):
|
||
tp = TrendPortfolio(**CANONICAL)
|
||
df = resample_tf(load(asset, "1h"), tf)
|
||
net, ts = tp.net_returns(df)
|
||
return pd.Series(net, index=pd.to_datetime(ts.values))
|
||
|
||
|
||
def corr_to_tp01(net, idx, tp_series):
|
||
s = pd.Series(net, index=idx)
|
||
j = pd.concat([s.rename("a"), tp_series.rename("b")], axis=1, join="inner").fillna(0.0)
|
||
if j["a"].std() == 0 or j["b"].std() == 0:
|
||
return 0.0
|
||
return float(j["a"].corr(j["b"]))
|
||
|
||
|
||
# ===========================================================================
|
||
# Reports
|
||
# ===========================================================================
|
||
def report_headline(tf, quick):
|
||
print("\n" + "=" * 120)
|
||
print(f"# HEADLINE — TF {tf} | standalone signals, full / IS / OOS, turnover, corr→TP01 (fee 0.10% RT)")
|
||
print("=" * 120)
|
||
tp = {a: tp01_net(a, tf) for a in ASSETS}
|
||
print(f" {'signal':<14s}{'asset':<6s}"
|
||
f"{'fullShrp':>9s}{'fullCAGR':>9s}{'fullDD':>7s}"
|
||
f"{'IS_Shrp':>8s}{'OOS_Shrp':>9s}{'OOS_ret':>8s}{'turn/y':>8s}{'corrTP':>8s}")
|
||
results = {}
|
||
for name, (fn, params) in SIGNALS.items():
|
||
for a in ASSETS:
|
||
res = eval_signal(fn, params, tf, a)
|
||
cr = corr_to_tp01(res["net"], res["idx"], tp[a])
|
||
results[(name, a)] = (res, cr)
|
||
print(f" {name:<14s}{a:<6s}"
|
||
f"{res['full']['sharpe']:>9.2f}{res['full']['cagr']*100:>8.1f}%"
|
||
f"{res['full']['max_dd']*100:>6.1f}%"
|
||
f"{res['is_']['sharpe']:>8.2f}{res['oos']['sharpe']:>9.2f}"
|
||
f"{res['oos']['total']*100:>7.1f}%{res['full']['ann_turnover']:>8.1f}{cr:>8.2f}")
|
||
return results, tp
|
||
|
||
|
||
def report_peryear(results):
|
||
print("\n" + "-" * 120)
|
||
print("# PER-YEAR net return (%) — only signals with OOS Sharpe>0 on BOTH assets shown")
|
||
print("-" * 120)
|
||
years = list(range(2018, 2027))
|
||
# which signals pass OOS>0 both assets
|
||
good = []
|
||
for name in SIGNALS:
|
||
if all(results[(name, a)][0]["oos"]["sharpe"] > 0 for a in ASSETS):
|
||
good.append(name)
|
||
if not good:
|
||
print(" (none — no signal has positive OOS Sharpe on BOTH assets)")
|
||
return good
|
||
print(" " + " " * 22 + "".join(f"{y:>7d}" for y in years))
|
||
for name in good:
|
||
for a in ASSETS:
|
||
py = results[(name, a)][0]["py"]
|
||
row = "".join((" . " if y not in py else f"{py[y]*100:>+7.0f}") for y in years)
|
||
print(f" {name+' '+a:<22s}{row}")
|
||
return good
|
||
|
||
|
||
def report_grid(quick):
|
||
print("\n" + "=" * 120)
|
||
print("# GRID ROBUSTNESS (TF 12h) — fraction of cells with positive full+OOS Sharpe on BOTH assets")
|
||
print("=" * 120)
|
||
tf = "12h"
|
||
grids = {
|
||
"VolBreakout": ("sig", sig_vol_breakout,
|
||
dict(don=[10, 20, 40] if not quick else [20],
|
||
zwin=[10, 20, 40], zk=[0.5, 1.0, 2.0])),
|
||
"OBV-trend": ("sig", sig_obv_trend, dict(ma=[15, 30, 60, 100])),
|
||
"VW-mom": ("sig", sig_vw_momentum,
|
||
dict(mom_win=[15, 30, 60, 90], long_only=[True])),
|
||
"RangeExpand": ("sig", sig_range_expansion,
|
||
dict(rng_win=[10, 20, 40], k=[1.3, 1.5, 2.0], hold=[3, 5, 10])),
|
||
"VT-long": ("sig", sig_vt_long, dict(target_vol=[0.15, 0.20, 0.30],
|
||
vol_win_days=[15, 30, 60])),
|
||
}
|
||
from itertools import product
|
||
for name, (_, fn, axes) in grids.items():
|
||
keys = list(axes.keys())
|
||
combos = list(product(*[axes[k] for k in keys]))
|
||
npos = 0
|
||
best = (-9, None)
|
||
for combo in combos:
|
||
params = dict(zip(keys, combo))
|
||
ok = True
|
||
sh_sum = 0.0
|
||
for a in ASSETS:
|
||
res = eval_signal(fn, params, tf, a)
|
||
if not (res["full"]["sharpe"] > 0 and res["oos"]["sharpe"] > 0):
|
||
ok = False
|
||
sh_sum += res["oos"]["sharpe"]
|
||
if ok:
|
||
npos += 1
|
||
if sh_sum > best[0]:
|
||
best = (sh_sum, params)
|
||
print(f" {name:<14s} positive both(full&OOS): {npos:>3d}/{len(combos):<3d} "
|
||
f"({npos/len(combos)*100:>4.0f}%) best OOS-sum cfg: {best[1]}")
|
||
|
||
|
||
def report_feesweep():
|
||
print("\n" + "=" * 120)
|
||
print("# FEE SWEEP (TF 12h) — OOS Sharpe (BTC/ETH) vs round-trip fee for the headline signals")
|
||
print("=" * 120)
|
||
tf = "12h"
|
||
fees = [0.0, 0.0005, 0.001, 0.0015, 0.002] # per side; RT = 2x
|
||
print(f" {'signal':<14s}" + "".join(f" RT{f*2*100:>4.2f}%" for f in fees))
|
||
for name, (fn, params) in SIGNALS.items():
|
||
cells = []
|
||
for f in fees:
|
||
shs = []
|
||
for a in ASSETS:
|
||
res = eval_signal(fn, params, tf, a, fee_side=f)
|
||
shs.append(res["oos"]["sharpe"])
|
||
cells.append(f"{shs[0]:>4.2f}/{shs[1]:<4.2f}")
|
||
print(f" {name:<14s}" + "".join(f" {c:>9s}" for c in cells))
|
||
|
||
|
||
# ===========================================================================
|
||
# REGIME FILTER on TP01 — does a vol/volume regime mask lift Sharpe or cut DD?
|
||
# ===========================================================================
|
||
def vol_regime_mask(df, bpd, win_days=30, mode="low", q=0.5):
|
||
"""Boolean per-bar mask (decided <= close[i]) for a realized-vol regime.
|
||
mode='low': keep exposure when vol <= rolling median; 'high': when vol > median."""
|
||
c = df["close"].values.astype(float)
|
||
r = simple_returns(c)
|
||
bpy = bpd * 365.25
|
||
vol = realized_vol(r, win_days * bpd, bpy)
|
||
# causal expanding/rolling quantile threshold (use a long rolling window, prior bars)
|
||
thr = pd.Series(vol).shift(1).rolling(180 * bpd, min_periods=30 * bpd).quantile(q).values
|
||
if mode == "low":
|
||
mask = vol <= thr
|
||
else:
|
||
mask = vol > thr
|
||
return np.nan_to_num(mask.astype(float), nan=1.0) # default keep before warmup
|
||
|
||
|
||
def vol_managed_mask(df, bpd, win_days=30, target_vol=0.20, cap=1.5):
|
||
"""Continuous vol-scaling multiplier on TP01: scale exposure by target_vol/realized_vol,
|
||
capped — an explicit volatility-managed overlay distinct from TP01's own sizing."""
|
||
c = df["close"].values.astype(float)
|
||
r = simple_returns(c)
|
||
bpy = bpd * 365.25
|
||
vol = realized_vol(r, win_days * bpd, bpy)
|
||
mult = np.where((vol > 0) & np.isfinite(vol), target_vol / vol, 1.0)
|
||
return np.clip(mult, 0.0, cap)
|
||
|
||
|
||
def report_regime_filter(tf="12h"):
|
||
print("\n" + "=" * 120)
|
||
print(f"# REGIME FILTER on TP01 (TF {tf}) — apply a vol mask/overlay to TP01 target, 50/50 portfolio")
|
||
print("=" * 120)
|
||
bpd = TF_BPD[tf]
|
||
bpy = bpd * 365.25
|
||
tp = TrendPortfolio(**CANONICAL)
|
||
|
||
def portfolio(transform):
|
||
"""transform(df,target)->target'; returns combined 50/50 net series + idx."""
|
||
series = {}
|
||
for a in ASSETS:
|
||
df = resample_tf(load(a, "1h"), tf)
|
||
r = simple_returns(df["close"].values.astype(float))
|
||
tgt = tp.target_series(df)
|
||
tgt2 = transform(df, tgt)
|
||
net, _, _ = net_from_target(tgt2, r, CANONICAL["fee_side"])
|
||
series[a] = pd.Series(net, index=pd.to_datetime(df["datetime"].values))
|
||
J = pd.concat(series, axis=1, join="inner").fillna(0.0)
|
||
combo = 0.5 * J[ASSETS[0]].values + 0.5 * J[ASSETS[1]].values
|
||
return combo, J.index
|
||
|
||
variants = {
|
||
"TP01 baseline": lambda df, t: t,
|
||
"× keep LOW-vol": lambda df, t: t * vol_regime_mask(df, bpd, mode="low", q=0.5),
|
||
"× keep HIGH-vol": lambda df, t: t * vol_regime_mask(df, bpd, mode="high", q=0.5),
|
||
"× keep LOW-vol q.7": lambda df, t: t * vol_regime_mask(df, bpd, mode="low", q=0.7),
|
||
"× vol-managed x1.5": lambda df, t: t * vol_managed_mask(df, bpd, cap=1.5) /
|
||
np.maximum(vol_managed_mask(df, bpd, cap=1.5).mean(), 1e-9),
|
||
"× obv-up only": lambda df, t: t * (np.where(
|
||
np.cumsum(np.sign(simple_returns(df['close'].values.astype(float))) * df['volume'].values)
|
||
> pd.Series(np.cumsum(np.sign(simple_returns(df['close'].values.astype(float)))
|
||
* df['volume'].values)).ewm(span=30, adjust=False).mean().values,
|
||
1.0, 0.0)),
|
||
}
|
||
print(f" {'variant':<22s}{'fullShrp':>9s}{'IS_Shrp':>8s}{'OOS_Shrp':>9s}"
|
||
f"{'CAGR':>8s}{'maxDD':>8s}{'turn/y':>9s}")
|
||
for name, tr in variants.items():
|
||
combo, idx = portfolio(tr)
|
||
m = metrics(combo, idx, np.zeros_like(combo), bpy)
|
||
cut = int(len(combo) * OOS_FRAC)
|
||
mi = metrics(combo[:cut], idx[:cut], np.zeros_like(combo[:cut]), bpy)
|
||
mo = metrics(combo[cut:], idx[cut:], np.zeros_like(combo[cut:]), bpy)
|
||
tt = 0.0
|
||
for a in ASSETS:
|
||
df = resample_tf(load(a, "1h"), tf)
|
||
tgt2 = tr(df, tp.target_series(df))
|
||
tt += np.sum(np.abs(np.diff(np.nan_to_num(tgt2), prepend=0.0)))
|
||
ann_tt = tt / m["years"] / 2.0
|
||
print(f" {name:<22s}{m['sharpe']:>9.2f}{mi['sharpe']:>8.2f}{mo['sharpe']:>9.2f}"
|
||
f"{m['cagr']*100:>7.1f}%{m['max_dd']*100:>7.1f}%{ann_tt:>9.1f}")
|
||
|
||
# robustness of the OBV-up filter across EMA spans (is 1.49 luck or stable?)
|
||
print("\n OBV-up filter robustness across EMA span (full / OOS Sharpe, maxDD):")
|
||
for span in [15, 20, 30, 45, 60, 90]:
|
||
def tr(df, t, sp=span):
|
||
c = df['close'].values.astype(float)
|
||
v = df['volume'].values.astype(float)
|
||
obv = np.cumsum(np.sign(simple_returns(c)) * v)
|
||
ema = pd.Series(obv).ewm(span=sp, adjust=False).mean().values
|
||
return t * np.where(obv > ema, 1.0, 0.0)
|
||
combo, idx = portfolio(tr)
|
||
m = metrics(combo, idx, np.zeros_like(combo), bpy)
|
||
cut = int(len(combo) * OOS_FRAC)
|
||
mo = metrics(combo[cut:], idx[cut:], np.zeros_like(combo[cut:]), bpy)
|
||
py = per_year(combo, idx)
|
||
neg_years = sum(1 for y, v in py.items() if v < 0)
|
||
print(f" span {span:>3d}: full {m['sharpe']:>4.2f} OOS {mo['sharpe']:>4.2f} "
|
||
f"DD {m['max_dd']*100:>4.1f}% CAGR {m['cagr']*100:>5.1f}% neg-years {neg_years}")
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--quick", action="store_true")
|
||
args = ap.parse_args()
|
||
|
||
print("#" * 120)
|
||
print("# TRACK H — VOLUME / RANGE / VOLATILITY-REGIME on certified BTC/ETH (Deribit mainnet)")
|
||
print("# Honest engine: target decided <=close[i], held bar i+1; fee on |Δpos|; OOS 65/35; >=12h only.")
|
||
print("#" * 120)
|
||
|
||
tfs = ["12h"] if args.quick else ["12h", "1d"]
|
||
for tf in tfs:
|
||
results, tp = report_headline(tf, args.quick)
|
||
report_peryear(results)
|
||
if tf == "12h":
|
||
crosscheck_backtest_signals()
|
||
report_grid(args.quick)
|
||
report_feesweep()
|
||
report_regime_filter("12h")
|
||
|
||
print("\n" + "#" * 120)
|
||
print("# VERDICT (track H) — honest reading of the tables above")
|
||
print("#" * 120)
|
||
for line in [
|
||
"1. NO uncorrelated additive edge found. Every PROFITABLE volume/range/vol signal",
|
||
" (VolBreakout, OBV-trend, VW-mom, VT-long) is trend-in-disguise: corr-to-TP01 0.61-0.75.",
|
||
" They do not diversify TP01 -> cannot raise the 50/50 portfolio Sharpe.",
|
||
"2. The genuinely LOWER-corr signals (RangeExpand ~0.48, NR7 ~0.48) FAIL OOS on >=1 asset",
|
||
" (NR7 ETH OOS Sharpe ~0.0/-0.03; RangeExpand BTC weak, ETH negative on 1d). Not deployable.",
|
||
"3. Declining-volume / fade (mean-reversion) is firmly NEGATIVE net of fees on both assets",
|
||
" and at ZERO fee -> confirms the v2.0.0 lesson: MR edge was feed contamination, it is dead.",
|
||
"4. Vol-REGIME gating of TP01 (keep low-vol / keep high-vol) HURTS Sharpe (1.32 -> 0.94/0.98).",
|
||
" A vol-managed x1.5 overlay leaves Sharpe ~flat (1.33) but raises DD (17.9%). No win.",
|
||
"5. The ONLY non-harmful overlay is an OBV-up trend-CONFIRMATION filter (keep TP01 long only",
|
||
" while OBV>EMA): full Sharpe 1.32->1.49, maxDD 13.3%->10.1%, but CAGR 16.2%->14.4%, turnover",
|
||
" +60%, and OOS gain is marginal (0.90->1.04) and span-sensitive (fades for EMA>45). It is",
|
||
" trend double-confirmation (de-risking), NOT new alpha. Worth noting as a DEFENSIVE overlay",
|
||
" if cutting DD matters more than CAGR; it does NOT robustly raise the portfolio Sharpe.",
|
||
"BOTTOM LINE: the ~1.3 portfolio-Sharpe ceiling on BTC/ETH-only HOLDS. Volume/range/vol add",
|
||
"nothing uncorrelated. TP01 stays the deployable winner.",
|
||
]:
|
||
print(" " + line)
|
||
print("#" * 120)
|
||
|
||
|
||
def crosscheck_backtest_signals():
|
||
"""Cross-check two DISCRETE signals through the canonical harness `backtest_signals`
|
||
(decide<=close[i], fill at close[i]) to confirm the per-bar engine isn't flattering them."""
|
||
print("\n" + "-" * 120)
|
||
print("# CROSS-CHECK via harness.backtest_signals (discrete entries, fee 0.10% RT, TF 12h)")
|
||
print("-" * 120)
|
||
tf = "12h"
|
||
for a in ASSETS:
|
||
df = resample_tf(load(a, "1h"), tf)
|
||
h = df["high"].values.astype(float)
|
||
l = df["low"].values.astype(float)
|
||
c = df["close"].values.astype(float)
|
||
rng = h - l
|
||
avg = roll_mean_prior(rng, 20)
|
||
pos_in_bar = np.where(rng > 0, (c - l) / rng, 0.5)
|
||
expand = rng > 1.5 * avg
|
||
entries = [None] * len(df)
|
||
for i in range(len(df)):
|
||
if expand[i] and pos_in_bar[i] > 0.6:
|
||
entries[i] = dict(dir=1, tp=None, sl=None, max_bars=5)
|
||
m = backtest_signals(df, entries, fee_rt=0.001, leverage=1.0, asset=a, tf=tf)
|
||
m.print_summary(f"RangeExpand(L,5b) {a}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|