"""TRACK A — TREND / MOMENTUM research on certified BTC/ETH (Deribit mainnet). Honest harness only (src.backtest.harness). Rules enforced: * Direction & entry price decided with data <= close[i]; fill at close[i]. * Net of fees (0.10% RT baseline) + fee sweep + leverage stress. * IS / OOS split (65/35). Grid robustness across params AND both assets. Run: uv run python scripts/research/trackA_trend.py This script is deliberately skeptical: it prints full grids so the reader can see whether an "edge" is a single lucky cell or a robust neighborhood. The verdict at the end is printed from the actual numbers, not asserted. """ from __future__ import annotations 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, oos_split ASSETS = ["BTC", "ETH"] TFS = ["1h", "15m", "5m"] FEE = 0.001 # --------------------------------------------------------------------------- # Signal builders. Each returns a list[dict|None] of length len(df). # All features use ONLY data up to and including close[i]. Entry fills at close[i]. # Position is approximated as a chained, non-overlapping hold of `hold` bars whose # direction is recomputed at each (free) bar -> amortizes fee over `hold` bars while # staying honest about responsiveness. # --------------------------------------------------------------------------- def sig_tsmom(df, lookback, hold, long_only=False): c = df["close"].values n = len(c) ent = [None] * n dirs = np.where(c[lookback:] > c[:-lookback], 1, -1) for k, d in enumerate(dirs): if long_only and d < 0: continue ent[lookback + k] = {"dir": int(d), "max_bars": hold} return ent def _ema(x, span): return pd.Series(x).ewm(span=span, adjust=False).mean().values def sig_ema_cross(df, fast, slow, hold, long_only=False): c = df["close"].values n = len(c) ef = _ema(c, fast) es = _ema(c, slow) ent = [None] * n for i in range(slow, n): d = 1 if ef[i] > es[i] else -1 if long_only and d < 0: ent[i] = None continue ent[i] = {"dir": d, "max_bars": hold} return ent def sig_donchian(df, lookback, hold, long_only=False): """Breakout: close[i] strictly above prior `lookback` highs -> long; below lows -> short. Detection AND entry both at close[i] (honest).""" c = df["close"].values h = df["high"].values l = df["low"].values n = len(c) ent = [None] * n # prior-window high/low EXCLUDING current bar (shift by 1) -> honest hh = pd.Series(h).rolling(lookback).max().shift(1).values ll = pd.Series(l).rolling(lookback).min().shift(1).values for i in range(lookback, n): if not np.isfinite(hh[i]): continue if c[i] > hh[i]: d = 1 elif c[i] < ll[i]: d = -1 else: continue if long_only and d < 0: continue ent[i] = {"dir": d, "max_bars": hold} return ent def sig_vol_scaled_tsmom(df, lookback, hold, vol_win, z_gate): """Momentum gated by trend strength: only take a position when |past return| exceeds z_gate * rolling stdev of bar returns (regime gate). Honest: all <= close[i].""" c = df["close"].values n = len(c) logret = np.zeros(n) logret[1:] = np.diff(np.log(c)) vol = pd.Series(logret).rolling(vol_win).std().values ent = [None] * n start = max(lookback, vol_win) + 1 for i in range(start, n): r = np.log(c[i] / c[i - lookback]) v = vol[i] * np.sqrt(lookback) if not np.isfinite(v) or v == 0: continue z = r / v if abs(z) < z_gate: continue d = 1 if z > 0 else -1 ent[i] = {"dir": d, "max_bars": hold} return ent # --------------------------------------------------------------------------- # Evaluation helpers # --------------------------------------------------------------------------- def eval_is_oos(df, entries, asset, tf, fee=FEE, lev=1.0): cut = oos_split(df, 0.65) full = backtest_signals(df, entries, fee_rt=fee, leverage=lev, asset=asset, tf=tf) ent_is = [e if i < cut else None for i, e in enumerate(entries)] ent_oos = [e if i >= cut else None for i, e in enumerate(entries)] m_is = backtest_signals(df, ent_is, fee_rt=fee, leverage=lev, asset=asset, tf=tf) m_oos = backtest_signals(df, ent_oos, fee_rt=fee, leverage=lev, asset=asset, tf=tf) return full, m_is, m_oos def buy_hold(df, cut=None): c = df["close"].values if cut is None: cut = oos_split(df, 0.65) return c[-1] / c[0] - 1, c[-1] / c[cut] - 1 # (full, oos) def print_benchmarks(): print("\n" + "=" * 110) print("# BUY & HOLD BENCHMARK (the bar any long/short trend edge must clear)") print("# NOTE: OOS window is the LAST 35% = ~late-2023 -> 2026, a single (mostly bull) regime.") print("# 2018-2022 (bear+crash+bull+bear) is ENTIRELY in-sample. 'positive OOS' is weak evidence.") print("=" * 110) for tf in TFS: for asset in ASSETS: df = load(asset, tf) cut = oos_split(df, 0.65) bf, bo = buy_hold(df, cut) print(f" {asset} {tf:>3s} OOS starts {df['datetime'].iloc[cut].date()} " f"B&H full={bf*100:>+7.0f}% B&H OOS={bo*100:>+7.0f}%") def line(label, m): print(f" {label:<30s} tr={m.n_trades:>6d} wr={m.win_rate:>4.1f}% " f"ret={m.net_return*100:>+8.0f}% CAGR={m.cagr*100:>+6.1f}% " f"Sh={m.sharpe:>5.2f} DD={m.max_dd*100:>4.1f}% mkt={m.time_in_market*100:>3.0f}% " f"€/d={m.daily_profit(2000):>+6.2f}") # --------------------------------------------------------------------------- # Experiments # --------------------------------------------------------------------------- def run_grid(name, builder, param_grid, builder_kwargs_fn, tfs=TFS, assets=ASSETS): """Generic grid runner. Prints OOS-focused table. Returns list of result dicts.""" print("\n" + "=" * 110) print(f"# {name}") print("=" * 110) results = [] for tf in tfs: for asset in assets: df = load(asset, tf) print(f"\n -- {asset} {tf} (n={len(df)}) --") for params in param_grid: ent = builder(df, **builder_kwargs_fn(params)) full, m_is, m_oos = eval_is_oos(df, ent, asset, tf) tag = ",".join(f"{k}={v}" for k, v in params.items()) line(f"{tag} [OOS]", m_oos) results.append(dict(name=name, asset=asset, tf=tf, params=params, full=full, is_=m_is, oos=m_oos)) return results def summarize_survivors(all_results): print("\n" + "#" * 110) print("# SURVIVOR SCREEN — positive OOS net return AND positive full-sample, Sharpe(OOS)>0") print("#" * 110) survivors = [r for r in all_results if r["oos"].net_return > 0 and r["full"].net_return > 0 and r["oos"].sharpe > 0 and r["oos"].n_trades >= 20] if not survivors: print(" NONE. No config is net-positive OOS with positive full-sample and Sharpe>0.") return [] survivors.sort(key=lambda r: r["oos"].sharpe, reverse=True) # precompute B&H OOS per (asset,tf) bh = {} for tf in TFS: for a in ASSETS: bh[(a, tf)] = buy_hold(load(a, tf))[1] print(" (BEATS B&H = OOS return exceeds buy&hold over same OOS window; otherwise it's just beta)") for r in survivors[:40]: tag = ",".join(f"{k}={v}" for k, v in r["params"].items()) bho = bh[(r["asset"], r["tf"])] beat = "BEATS B&H" if r["oos"].net_return > bho else "<= B&H (beta)" print(f" {r['name'][:18]:<18s} {r['asset']} {r['tf']:>3s} {tag:<28s} " f"OOS: ret={r['oos'].net_return*100:>+7.0f}% Sh={r['oos'].sharpe:>4.2f} " f"DD={r['oos'].max_dd*100:>4.0f}% €/d={r['oos'].daily_profit(2000):>+5.2f} | " f"B&H={bho*100:>+5.0f}% {beat}") return survivors def robustness_report(survivors): """For top survivors, check fee sweep + leverage stress + cross-asset consistency.""" if not survivors: return print("\n" + "#" * 110) print("# ROBUSTNESS: fee sweep (0.0005/0.001/0.0015/0.002) + leverage (1x/2x/3x) on top survivors") print("#" * 110) seen = set() for r in survivors[:8]: key = (r["name"], r["asset"], r["tf"], tuple(r["params"].items())) if key in seen: continue seen.add(key) df = load(r["asset"], r["tf"]) # rebuild entries builder = BUILDERS[r["name"]] ent = builder(df, **KW_FN[r["name"]](r["params"])) tag = ",".join(f"{k}={v}" for k, v in r["params"].items()) print(f"\n {r['name']} {r['asset']} {r['tf']} {tag}") print(" fee sweep (OOS net return):") for fee in (0.0005, 0.001, 0.0015, 0.002): _, _, m_oos = eval_is_oos(df, ent, r["asset"], r["tf"], fee=fee) flag = "" if m_oos.net_return > 0 else " <-- DIES" print(f" fee={fee:.4f}: OOS ret={m_oos.net_return*100:>+8.0f}% Sh={m_oos.sharpe:>4.2f}{flag}") print(" leverage stress (OOS, fee=0.001):") for lev in (1.0, 2.0, 3.0): _, _, m_oos = eval_is_oos(df, ent, r["asset"], r["tf"], lev=lev) print(f" {lev:.0f}x: OOS ret={m_oos.net_return*100:>+8.0f}% " f"Sh={m_oos.sharpe:>4.2f} DD={m_oos.max_dd*100:>4.0f}% €/d={m_oos.daily_profit(2000):>+5.2f}") # yearly OOS _, _, m_oos = eval_is_oos(df, ent, r["asset"], r["tf"]) print(" OOS yearly:") for y in sorted(m_oos.yearly): print(f" {y}: {m_oos.yearly[y]*100:>+7.1f}%") # registry so robustness_report can rebuild entries BUILDERS = { "TSMOM": sig_tsmom, "TSMOM_LONG": sig_tsmom, "EMA_CROSS": sig_ema_cross, "DONCHIAN": sig_donchian, "VOLSCALED_TSMOM": sig_vol_scaled_tsmom, } KW_FN = { "TSMOM": lambda p: dict(lookback=p["N"], hold=p["H"]), "TSMOM_LONG": lambda p: dict(lookback=p["N"], hold=p["H"], long_only=True), "EMA_CROSS": lambda p: dict(fast=p["f"], slow=p["s"], hold=p["H"]), "DONCHIAN": lambda p: dict(lookback=p["N"], hold=p["H"]), "VOLSCALED_TSMOM": lambda p: dict(lookback=p["N"], hold=p["H"], vol_win=p["vw"], z_gate=p["z"]), } def main(): pd.set_option("display.width", 200) print_benchmarks() all_results = [] # ---- 1. TSMOM (long/short) ---- tsmom_grid = [dict(N=n, H=h) for n in (10, 20, 50, 100, 200) for h in (6, 12, 24, 48)] all_results += run_grid("TSMOM", sig_tsmom, tsmom_grid, KW_FN["TSMOM"]) # ---- 2. TSMOM long-only (crypto has strong upward drift; honest to test) ---- all_results += run_grid("TSMOM_LONG", lambda df, **k: sig_tsmom(df, long_only=True, **k), [dict(N=n, H=h) for n in (20, 50, 100, 200) for h in (12, 24, 48)], KW_FN["TSMOM"]) # ---- 3. EMA crossover ---- ema_grid = [dict(f=f, s=s, H=h) for (f, s) in ((10, 30), (20, 50), (20, 100), (50, 200)) for h in (12, 24, 48)] all_results += run_grid("EMA_CROSS", sig_ema_cross, ema_grid, KW_FN["EMA_CROSS"]) # ---- 4. Donchian breakout ---- don_grid = [dict(N=n, H=h) for n in (20, 50, 100, 200) for h in (12, 24, 48)] all_results += run_grid("DONCHIAN", sig_donchian, don_grid, KW_FN["DONCHIAN"]) # ---- 5. Vol-scaled / regime-gated TSMOM ---- vs_grid = [dict(N=n, H=h, vw=vw, z=z) for n in (20, 50, 100) for h in (24, 48) for vw in (50, 100) for z in (0.5, 1.0)] all_results += run_grid("VOLSCALED_TSMOM", sig_vol_scaled_tsmom, vs_grid, KW_FN["VOLSCALED_TSMOM"]) # ---- survivor screen + robustness ---- survivors = summarize_survivors(all_results) robustness_report(survivors) # ---- cross-asset robustness note ---- print("\n" + "#" * 110) print("# CROSS-ASSET / CROSS-TF CONSISTENCY of survivors (a real edge holds on BOTH BTC & ETH)") print("#" * 110) from collections import defaultdict by_strat = defaultdict(list) for r in survivors: by_strat[(r["name"], r["tf"], tuple(r["params"].items()))].append(r["asset"]) both = [(k, v) for k, v in by_strat.items() if set(v) >= {"BTC", "ETH"}] if not both: print(" No single (strategy, tf, params) cell is an OOS survivor on BOTH BTC and ETH.") print(" => any apparent edge is asset/regime-specific, not a robust trend edge.") else: for (name, tf, params), assets in both: print(f" {name} {tf} {dict(params)} survives on: {assets}") print("\nDONE. Read the survivor screen + robustness above for the honest verdict.") if __name__ == "__main__": main()