Adriano/PythagorasGoal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
dc2b5697dab59cab45c359ba747236a561501763
PythagorasGoal/scripts/research/trackC_meanrev.py
T
Adriano dc2b5697da research wave 1: 5 honest tracks on certified BTC/ETH + synthesis 
- trackA trend, trackB ML, trackC mean-rev, trackD trend-portfolio, trackE xsec/ensemble
- VERDICT: Track D vol-targeted BTC+ETH trend portfolio is the one robust deployable
  earner (Sharpe 1.0-1.32, DD 13-19%, positive every year 2019-2026)
- mean-reversion confirmed dead on clean data; weak-but-real ML/trend residuals
- honest: EUR50/day on 2000 in 1-2y is not reachable (needs ~137k capital or ruinous DD)
2026-06-19 19:14:53 +02:00

381 lines
17 KiB
Python
Raw Blame History

"""TRACK C — Mean-reversion / range re-examination on CLEAN BTC/ETH (Deribit mainnet).
HONEST harness only. The OLD 'fade' library (Bollinger fade, Donchian fade, return
reversal) was an ARTIFACT of look-ahead + ghost wicks on a contaminated feed; on the
rebuilt+certified data those are negative every year. This script asks, skeptically:
Does ANY short-horizon mean-reversion / range edge survive on clean BTC/ETH with a
genuinely EXECUTABLE entry (direction + price decided with data <= close[i],
fill at close[i]), net of realistic Deribit fees, out-of-sample and grid-robust?
Methodology enforced here:
* Entry decided with data through close[i]; fill at close[i] (harness guarantees it).
No entering "at the band edge" / candle extreme only known intrabar.
* NET fees fee_rt=0.001 baseline + sweep {0.0005, 0.0015, 0.002}.
* OOS 65/35 split + parameter grid across BOTH BTC & ETH.
* Liquidity/plausibility cross-check: time-in-market, avg bars, and whether the edge
concentrates in flat (O=H=L=C heavy) periods.
Run:
uv run python scripts/research/trackC_meanrev.py # full (slow, all TFs)
uv run python scripts/research/trackC_meanrev.py --quick # 1h + 15m only
"""
from __future__ import annotations
import argparse
import sys
import time
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, Metrics
# ===========================================================================
# Indicator helpers — ALL causal: value at index i uses ONLY data through i.
# ===========================================================================
def zscore(close: np.ndarray, lookback: int) -> np.ndarray:
s = pd.Series(close)
ma = s.rolling(lookback).mean()
sd = s.rolling(lookback).std(ddof=0)
z = (s - ma) / sd
return z.values, ma.values, sd.values
def rsi(close: np.ndarray, period: int) -> np.ndarray:
s = pd.Series(close)
d = s.diff()
up = d.clip(lower=0.0)
dn = (-d).clip(lower=0.0)
# Wilder smoothing via ewm alpha=1/period (causal)
ru = up.ewm(alpha=1.0 / period, adjust=False).mean()
rd = dn.ewm(alpha=1.0 / period, adjust=False).mean()
rs = ru / rd.replace(0, np.nan)
out = 100 - 100 / (1 + rs)
return out.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
# ===========================================================================
# Signal generators — each returns a list[dict|None] length len(df).
# Direction/levels decided strictly with data through close[i].
# ===========================================================================
def sig_zfade(df, lookback=20, z=2.0, tp_mode="mean", tp_atr=1.0, sl_atr=2.0,
max_bars=24, atr_p=14):
"""Bollinger / z-score fade. z<-thr -> long (reversion up); z>+thr -> short.
TP at the moving mean (tp_mode='mean') or at tp_atr*ATR toward the mean.
SL at sl_atr*ATR beyond entry. Entry at close[i]."""
c = df["close"].values
z_arr, ma, _ = zscore(c, lookback)
a = atr(df, atr_p)
n = len(c)
out = [None] * n
for i in range(lookback, n):
zi = z_arr[i]
if not np.isfinite(zi) or not np.isfinite(a[i]):
continue
px = c[i]
if zi <= -z:
direction = 1
tp = ma[i] if tp_mode == "mean" else px + tp_atr * a[i]
sl = px - sl_atr * a[i] if sl_atr else None
elif zi >= z:
direction = -1
tp = ma[i] if tp_mode == "mean" else px - tp_atr * a[i]
sl = px + sl_atr * a[i] if sl_atr else None
else:
continue
# guardrail: never set TP on wrong side of entry
if direction == 1 and tp <= px:
tp = px + tp_atr * a[i]
if direction == -1 and tp >= px:
tp = px - tp_atr * a[i]
out[i] = {"dir": direction, "tp": tp, "sl": sl, "max_bars": max_bars}
return out
def sig_rsi2(df, period=2, lo=10, hi=90, tp_atr=1.0, sl_atr=2.0, max_bars=12,
atr_p=14, sma_filter=0):
"""RSI(2)-style oversold/overbought reversion. RSI<lo -> long, RSI>hi -> short.
Optional trend filter: only long above SMA(sma_filter), only short below."""
c = df["close"].values
r = rsi(c, period)
a = atr(df, atr_p)
sma = pd.Series(c).rolling(sma_filter).mean().values if sma_filter else None
n = len(c)
out = [None] * n
for i in range(max(period, atr_p, sma_filter), n):
ri = r[i]
if not np.isfinite(ri) or not np.isfinite(a[i]):
continue
px = c[i]
if ri <= lo:
if sma is not None and not (px > sma[i]):
continue
out[i] = {"dir": 1, "tp": px + tp_atr * a[i],
"sl": px - sl_atr * a[i] if sl_atr else None, "max_bars": max_bars}
elif ri >= hi:
if sma is not None and not (px < sma[i]):
continue
out[i] = {"dir": -1, "tp": px - tp_atr * a[i],
"sl": px + sl_atr * a[i] if sl_atr else None, "max_bars": max_bars}
return out
def sig_retrev(df, ret_lb=1, thr_sigma=2.0, vol_lb=50, tp_atr=1.0, sl_atr=2.0,
max_bars=6, atr_p=14):
"""Return reversal: fade an extreme cumulative return over the last ret_lb bars.
Extreme = |ret| > thr_sigma * rolling std of that return. Entry at close[i]."""
c = df["close"].values
s = pd.Series(c)
ret = np.log(s / s.shift(ret_lb))
sd = ret.rolling(vol_lb).std(ddof=0)
a = atr(df, atr_p)
n = len(c)
out = [None] * n
rv = ret.values
sv = sd.values
for i in range(vol_lb + ret_lb, n):
if not np.isfinite(rv[i]) or not np.isfinite(sv[i]) or sv[i] == 0 or not np.isfinite(a[i]):
continue
z = rv[i] / sv[i]
px = c[i]
if z <= -thr_sigma:
out[i] = {"dir": 1, "tp": px + tp_atr * a[i],
"sl": px - sl_atr * a[i] if sl_atr else None, "max_bars": max_bars}
elif z >= thr_sigma:
out[i] = {"dir": -1, "tp": px - tp_atr * a[i],
"sl": px + sl_atr * a[i] if sl_atr else None, "max_bars": max_bars}
return out
def sig_vwap(df, sess_bars=24, thr=2.0, tp_atr=1.0, sl_atr=2.0, max_bars=12, atr_p=14):
"""Rolling-VWAP distance reversion. Distance in std-of-distance units over a
rolling session window. Far above VWAP -> short, far below -> long. Entry close[i]."""
c = df["close"].values
v = df["volume"].values.astype(float)
tp = (df["high"].values + df["low"].values + c) / 3.0
pv = pd.Series(tp * v)
vol = pd.Series(v)
vwap = (pv.rolling(sess_bars).sum() / vol.rolling(sess_bars).sum()).values
dist = pd.Series(c - vwap)
dsd = dist.rolling(sess_bars).std(ddof=0).values
a = atr(df, atr_p)
n = len(c)
out = [None] * n
for i in range(sess_bars * 2, n):
if not np.isfinite(vwap[i]) or not np.isfinite(dsd[i]) or dsd[i] == 0 or not np.isfinite(a[i]):
continue
z = (c[i] - vwap[i]) / dsd[i]
px = c[i]
if z <= -thr:
out[i] = {"dir": 1, "tp": px + tp_atr * a[i],
"sl": px - sl_atr * a[i] if sl_atr else None, "max_bars": max_bars}
elif z >= thr:
out[i] = {"dir": -1, "tp": px - tp_atr * a[i],
"sl": px + sl_atr * a[i] if sl_atr else None, "max_bars": max_bars}
return out
# ===========================================================================
# Evaluation utilities
# ===========================================================================
def flat_fraction(df: pd.DataFrame) -> float:
o, h, l, c = df["open"], df["high"], df["low"], df["close"]
return float(((h == l) & (o == c)).mean())
def run_split(df, sigfn, params, fee_rt=0.001, leverage=1.0):
"""Run full / IS / OOS for a single config. Returns (full, is_, oos)."""
entries = sigfn(df, **params)
cut = oos_split(df, 0.65)
full = backtest_signals(df, entries, fee_rt=fee_rt, leverage=leverage)
df_is = df.iloc[:cut].reset_index(drop=True)
df_oos = df.iloc[cut:].reset_index(drop=True)
is_ = backtest_signals(df_is, sigfn(df_is, **params), fee_rt=fee_rt, leverage=leverage)
oos = backtest_signals(df_oos, sigfn(df_oos, **params), fee_rt=fee_rt, leverage=leverage)
return full, is_, oos
def hdr(title):
print("\n" + "=" * 92)
print(title)
print("=" * 92)
# ===========================================================================
# Main
# ===========================================================================
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--quick", action="store_true", help="1h+15m only (skip slow 5m)")
args = ap.parse_args()
t0 = time.time()
tfs = ["1h", "15m"] if args.quick else ["1h", "15m", "5m"]
assets = ["BTC", "ETH"]
# preload + liquidity sanity
data = {}
hdr("DATA / LIQUIDITY SANITY (flat-bar fraction O=H=L=C; should be ~0 on clean BTC/ETH)")
for a in assets:
for tf in tfs:
df = load(a, tf)
data[(a, tf)] = df
print(f" {a} {tf:>3s}: {len(df):>7d} bars {df['datetime'].iloc[0].date()}→"
f"{df['datetime'].iloc[-1].date()} flat={flat_fraction(df)*100:5.2f}%")
# -------------------------------------------------------------------
# PASS 1 — broad screen per family on 1h, both assets (IS/OOS).
# -------------------------------------------------------------------
hdr("PASS 1 — FAMILY SCREEN on 1h (honest entry, fee_rt=0.001, lev=1). "
"Look for OOS>0 on BOTH assets.")
families = {
"ZFADE z2/mean ": (sig_zfade, dict(lookback=20, z=2.0, tp_mode="mean", sl_atr=2.0, max_bars=24)),
"ZFADE z2.5/atr": (sig_zfade, dict(lookback=20, z=2.5, tp_mode="atr", tp_atr=1.5, sl_atr=2.0, max_bars=24)),
"ZFADE z3/mean ": (sig_zfade, dict(lookback=40, z=3.0, tp_mode="mean", sl_atr=3.0, max_bars=48)),
"RSI2 10/90 ": (sig_rsi2, dict(period=2, lo=10, hi=90, tp_atr=1.0, sl_atr=2.0, max_bars=12)),
"RSI2 5/95 ": (sig_rsi2, dict(period=2, lo=5, hi=95, tp_atr=1.5, sl_atr=2.5, max_bars=12)),
"RSI2 +trend ": (sig_rsi2, dict(period=2, lo=10, hi=90, tp_atr=1.0, sl_atr=2.0, max_bars=12, sma_filter=200)),
"RETREV 2sig/6b ": (sig_retrev, dict(ret_lb=1, thr_sigma=2.0, tp_atr=1.0, sl_atr=2.0, max_bars=6)),
"RETREV 3sig/12b": (sig_retrev, dict(ret_lb=3, thr_sigma=3.0, tp_atr=1.5, sl_atr=2.5, max_bars=12)),
"VWAP 2/sess24": (sig_vwap, dict(sess_bars=24, thr=2.0, tp_atr=1.0, sl_atr=2.0, max_bars=12)),
}
for name, (fn, params) in families.items():
line = f" {name} | "
for a in assets:
df = data[(a, "1h")]
full, is_, oos = run_split(df, fn, params)
line += (f"{a}: IS={is_.net_return*100:>+6.0f}% OOS={oos.net_return*100:>+6.0f}% "
f"(tr={oos.n_trades:>4d} wr={oos.win_rate:>4.1f} shrp={oos.sharpe:>+4.1f} "
f"mkt={oos.time_in_market*100:>3.0f}% ab={oos.avg_bars:>4.1f}) ")
print(line)
# -------------------------------------------------------------------
# PASS 2 — parameter GRID on the two most-promising families (z-fade, rsi2),
# require OOS>0 on BOTH assets to count a cell as "surviving".
# -------------------------------------------------------------------
hdr("PASS 2 — GRID ROBUSTNESS (1h). A cell 'survives' only if OOS net>0 on BOTH BTC AND ETH.")
def grid(fn, base, sweep, tf="1h"):
keys = list(sweep.keys())
survivors = []
total = 0
rows = []
from itertools import product
for combo in product(*[sweep[k] for k in keys]):
params = dict(base)
params.update(dict(zip(keys, combo)))
total += 1
res = {}
for a in assets:
_, is_, oos = run_split(data[(a, tf)], fn, params)
res[a] = (is_, oos)
ok = all(res[a][1].net_return > 0 for a in assets)
both_oos = np.mean([res[a][1].net_return for a in assets]) * 100
rows.append((params, res, ok))
if ok:
survivors.append((params, res))
print(f" {fn.__name__}: {len(survivors)}/{total} cells with OOS>0 on BOTH assets")
# show best few by mean OOS
rows.sort(key=lambda r: np.mean([r[1][a][1].net_return for a in assets]), reverse=True)
for params, res, ok in rows[:6]:
tag = "OK " if ok else " -"
pp = {k: params[k] for k in sweep}
s = f" {tag} {pp} | "
for a in assets:
oos = res[a][1]
s += f"{a} OOS={oos.net_return*100:>+6.0f}% (wr={oos.win_rate:>4.1f} shrp={oos.sharpe:>+4.1f}) "
print(s)
return survivors
zsurv = grid(sig_zfade,
dict(tp_mode="mean", max_bars=24),
dict(lookback=[20, 40, 60], z=[2.0, 2.5, 3.0], sl_atr=[2.0, 3.0]))
rsurv = grid(sig_rsi2,
dict(period=2, tp_atr=1.0),
dict(lo=[5, 10, 15], hi=[85, 90, 95], sl_atr=[2.0, 3.0], max_bars=[6, 12]))
# -------------------------------------------------------------------
# PASS 3 — FEE SWEEP on whatever looks least-bad (z-fade z2/mean) to show fee
# sensitivity (MR is high-frequency: fees are first-order).
# -------------------------------------------------------------------
hdr("PASS 3 — FEE SWEEP (z-fade lookback=20 z=2 mean, 1h). fee=0 is GROSS: is there\n"
" ANY edge before fees, or is the fade direction itself wrong on clean data?")
fees = [0.0, 0.0005, 0.001, 0.0015, 0.002]
base = dict(lookback=20, z=2.0, tp_mode="mean", sl_atr=2.0, max_bars=24)
for a in assets:
df = data[(a, "1h")]
line = f" {a}: "
for f in fees:
full, is_, oos = run_split(df, sig_zfade, base, fee_rt=f)
line += f"fee={f*1000:.1f}bp→ full={full.net_return*100:>+6.0f}% OOS={oos.net_return*100:>+6.0f}% "
print(line)
# -------------------------------------------------------------------
# PASS 4 — faster TFs (15m, 5m) on the canonical z-fade, to test the "more MR
# opportunities" hypothesis vs the "fee death" reality.
# -------------------------------------------------------------------
hdr("PASS 4 — z-fade across timeframes (lookback=20 z=2 mean). Faster TF = more fees.")
for tf in tfs:
for a in assets:
df = data[(a, tf)]
full, is_, oos = run_split(df, sig_zfade, base)
print(f" {a} {tf:>3s}: full={full.net_return*100:>+7.0f}% IS={is_.net_return*100:>+7.0f}% "
f"OOS={oos.net_return*100:>+7.0f}% tr={full.n_trades:>5d} wr={full.win_rate:>4.1f}% "
f"shrp={full.sharpe:>+4.1f} mkt={full.time_in_market*100:>3.0f}% €/d={full.daily_profit(2000):>+5.2f}")
# -------------------------------------------------------------------
# PASS 5 — SESSION / overnight effect (UTC hour-of-day) on 1h returns.
# Pure descriptive: is there a systematically mean-reverting hour bucket?
# -------------------------------------------------------------------
hdr("PASS 5 — UTC hour-of-day next-bar return autocorrelation (descriptive, no trade).")
for a in assets:
df = data[(a, "1h")]
c = df["close"].values
ret = pd.Series(np.log(c[1:] / c[:-1])) # ret[k] = log(c[k+1]/c[k])
prev = ret.shift(1)
hours = df["datetime"].dt.hour.values[1:1 + len(ret)]
tmp = pd.DataFrame({"h": hours[:len(ret)], "r": ret.values, "p": prev.values}).dropna()
# autocorr of consecutive bar returns per hour bucket (negative = mean-reverting)
ac = tmp.groupby("h").apply(lambda g: g["r"].corr(g["p"]) if len(g) > 30 else np.nan)
worst = ac.nsmallest(3)
best = ac.nlargest(3)
print(f" {a}: most mean-reverting UTC hours (neg autocorr): "
+ ", ".join(f"{int(h)}h={v:+.3f}" for h, v in worst.items())
+ " | most trending: "
+ ", ".join(f"{int(h)}h={v:+.3f}" for h, v in best.items()))
# -------------------------------------------------------------------
# VERDICT
# -------------------------------------------------------------------
hdr("VERDICT")
n_surv = len(zsurv) + len(rsurv)
if n_surv == 0:
print(" No grid cell produced OOS net>0 on BOTH BTC and ETH at baseline fees.")
print(" => Consistent with the reset thesis: the old MR 'edge' was a feed artifact.")
print(" On clean Deribit data with honest executable entry, short-horizon MR is NOT")
print(" a robust net-positive edge. (See per-pass tables above for the evidence.)")
else:
print(f" {n_surv} grid cell(s) survived OOS>0 on both assets. Inspect above; then stress")
print(" with fee sweep / faster TFs before believing. Surviving configs:")
for params, res in (zsurv + rsurv):
ms = np.mean([res[a][1].net_return for a in assets]) * 100
print(f" {params} meanOOS={ms:+.0f}%")
print(f"\n (elapsed {time.time()-t0:.0f}s)")
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink