research: porta artefatti da strategy-research-calendar (tracks F-I + eval crypto_backtest + lead OPZIONI/VRP)
Dal branch parallelo strategy-research-calendar (continuazione della linea TP01). Porta su main il record di ricerca + la fondazione del lead opzioni (NIENTE blob dati, niente codice in conflitto): - Tracks F/G/H/I (seasonality/calendar, prior-levels, volume-vol, momentum-reversal): tutti NEGATIVI/spurii -> confermano il soffitto Sharpe ~1.3 su BTC/ETH direzionale (calendar = buy&hold travestito; mean-reversion morta anche a fee 0). Diari + script. - trackD_lookahead_audit.py: audit anti-look-ahead (stesso esito del nostro fix >=12h). - eval-crypto-backtest-options.md: valutazione strategia esterna crypto_backtest. Cross-valida TP01 (il loro sleeve spot 12h ~ TP01: due ricerche indipendenti, stessa conclusione). Identifica il LEAD: sleeve income OPZIONI (vendita put settimanali delta-0.28, VRP IV>RV), scorrelato ~0.22 al trend -> via per superare il soffitto ~1.3. - options_real_quote_check.py + cerbero-bite-mainnet-verified.md: VERIFICATO su QUOTE REALI Deribit mainnet (cerbero-bite/MCP = mainnet, bit-identico a ccxt.deribit). Premio reale (BID, con skew) = 1.29x il modellato -> il backtest SOTTOSTIMA il premio; il rischio vero e' la CODA (short-vol) + liquidita' di roll in stress, non la magnitudine. NB: lo sleeve opzioni e' un LEAD, NON deployato: prezzato da modello (BS su DVOL) + 1 snapshot in regime calmo. Serve validazione real-chain multi-regime + stress crash + paper su testnet prima di aggiungerlo al portafoglio. Portafoglio attivo invariato: TP01 70% + XS01 30%. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Adriano Dal Pastro
@@ -0,0 +1,602 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user