audit+fix: anti-look-ahead audit, migrate deployable config to >=12h
- trackD_lookahead_audit.py: relabel test (left==right, no labeling leak) + execution-lag stress -> our trend pipeline is CLEAN (4h Sharpe 1.36 robust to +1 bar lag, label-invariant) - ADOPT conservative conclusion: deploy at 12h (sub-12h: costs/overfit dominate, slight Sharpe bump unreliable). 12h: Sharpe 1.32, DD 13.3%, CAGR 16.2% ~ identical and robust - trend_portfolio: DEPLOY_TF=12h, resample_tf(rule); paper trader + tests on 12h - calendar research (NEGATIVE, both): trackF seasonality (spurious), trackG prior-levels (breakouts continue, fade dead; only long-drift survivor, redundant with TP01) - gitignore data/paper_trend runtime state
This commit is contained in:
@@ -33,7 +33,8 @@ PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
||||
sys.path.insert(0, str(PROJECT_ROOT))
|
||||
|
||||
from src.backtest.harness import load
|
||||
from src.strategies.trend_portfolio import TrendPortfolio, CANONICAL, resample_4h, simple_returns
|
||||
from src.strategies.trend_portfolio import (
|
||||
TrendPortfolio, CANONICAL, resample_tf, DEPLOY_TF, simple_returns)
|
||||
|
||||
STATE_DIR = PROJECT_ROOT / "data" / "paper_trend"
|
||||
STATE_FILE = STATE_DIR / "state.json"
|
||||
@@ -43,8 +44,8 @@ WEIGHT = 0.5
|
||||
INITIAL_CAPITAL = 2000.0
|
||||
|
||||
|
||||
def build_4h() -> dict[str, pd.DataFrame]:
|
||||
return {a: resample_4h(load(a, "1h")) for a in ASSETS}
|
||||
def build_bars() -> dict[str, pd.DataFrame]:
|
||||
return {a: resample_tf(load(a, "1h"), DEPLOY_TF) for a in ASSETS}
|
||||
|
||||
|
||||
def load_state() -> dict | None:
|
||||
@@ -144,10 +145,10 @@ def print_status(st: dict, dfs: dict):
|
||||
ret = cap / st["initial_capital"] - 1
|
||||
daily = (cap - st["initial_capital"]) / days if days > 0 else 0.0
|
||||
print("=" * 72)
|
||||
print(" PAPER TRADER — TP01 Trend Portfolio (PORT LF4h, 50/50 BTC+ETH, 4h)")
|
||||
print(f" PAPER TRADER — TP01 Trend Portfolio (PORT LF{DEPLOY_TF}, 50/50 BTC+ETH)")
|
||||
print("=" * 72)
|
||||
print(f" start {start:%Y-%m-%d %H:%M} UTC")
|
||||
print(f" last bar {last:%Y-%m-%d %H:%M} UTC ({days:.1f} giorni, {st['n_bars']} barre 4h)")
|
||||
print(f" last bar {last:%Y-%m-%d %H:%M} UTC ({days:.1f} giorni, {st['n_bars']} barre {DEPLOY_TF})")
|
||||
print(f" capitale {cap:,.2f} USDT (start {st['initial_capital']:,.0f})")
|
||||
print(f" ritorno {ret*100:+.2f}% | €/giorno {daily:+.2f} | maxDD {st['max_dd']*100:.1f}%")
|
||||
print(f" posizioni now { 'flat' if all(p==0 for p in st['positions'].values()) else '' }")
|
||||
@@ -168,7 +169,7 @@ def print_status(st: dict, dfs: dict):
|
||||
|
||||
def main():
|
||||
argv = sys.argv[1:]
|
||||
dfs = build_4h()
|
||||
dfs = build_bars()
|
||||
if "--reset" in argv:
|
||||
if STATE_FILE.exists():
|
||||
STATE_FILE.unlink()
|
||||
|
||||
@@ -0,0 +1,118 @@
|
||||
"""ADVERSARIAL LOOK-AHEAD / EXECUTION-LAG AUDIT of the trend portfolio across timeframes.
|
||||
|
||||
Motivation (2026-06-19): a look-ahead bug (ffill mixed-timeframe on open-labeled bars) can
|
||||
inflate sub-daily Sharpe (e.g. 4h to ~1.6 vs a real ~1.1). This audit stress-tests OUR pipeline:
|
||||
|
||||
1. EXECUTION LAG: standard book holds the position decided at close[i] during bar i+1.
|
||||
We re-run with an EXTRA bar of delay (held during i+2) — i.e. you cannot trade exactly at
|
||||
the close; there is one bar of slippage/latency. A genuine slow-trend edge barely moves; a
|
||||
timing artifact collapses. We sweep lag = 1 (standard) and 2 (conservative).
|
||||
2. RELABEL TEST: resample with label='left' (open-labeled, our default) vs label='right'
|
||||
(close-labeled). The realized Sharpe must be (near) identical; a large gap => the labeling
|
||||
leaks information.
|
||||
|
||||
Conclusion target: identify the timeframe below which costs + lag dominate (don't deploy there).
|
||||
|
||||
Run: uv run python scripts/research/trackD_lookahead_audit.py
|
||||
"""
|
||||
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
|
||||
from src.strategies.trend_portfolio import simple_returns, realized_vol, tsmom_blend
|
||||
|
||||
ASSETS = ["BTC", "ETH"]
|
||||
FEE_SIDE = 0.0005
|
||||
TARGET_VOL = 0.20
|
||||
LEVERAGE = 2.0
|
||||
LONG_ONLY = True
|
||||
TFS = {"4h": ("4h", 6), "6h": ("6h", 4), "8h": ("8h", 3), "12h": ("12h", 2), "1d": ("1D", 1)}
|
||||
|
||||
|
||||
def resample(df1h: pd.DataFrame, rule: str, label: str) -> pd.DataFrame:
|
||||
g = df1h.copy()
|
||||
idx = pd.to_datetime(g["timestamp"], unit="ms", utc=True)
|
||||
idx.name = "dt"
|
||||
g.index = idx
|
||||
out = g.resample(rule, label=label, closed="left").agg(
|
||||
{"open": "first", "high": "max", "low": "min", "close": "last", "volume": "sum"})
|
||||
out = out.dropna(subset=["open"])
|
||||
out["datetime"] = out.index
|
||||
return out.reset_index(drop=True)
|
||||
|
||||
|
||||
def target_series(c, bpd):
|
||||
bpy = bpd * 365.25
|
||||
r = simple_returns(c)
|
||||
vol = realized_vol(r, 30 * bpd, bpy)
|
||||
direction = np.clip(tsmom_blend(c, (30 * bpd, 90 * bpd, 180 * bpd)), 0, None) if LONG_ONLY \
|
||||
else tsmom_blend(c, (30 * bpd, 90 * bpd, 180 * bpd))
|
||||
scal = np.where((vol > 0) & np.isfinite(vol), TARGET_VOL / vol, 0.0)
|
||||
tgt = np.clip(direction * scal, -LEVERAGE, LEVERAGE)
|
||||
tgt[~np.isfinite(tgt)] = 0.0
|
||||
return tgt, r
|
||||
|
||||
|
||||
def sleeve_net(df, bpd, lag):
|
||||
"""net[t] uses position decided at close[t-lag] (lag>=1). lag=1 = standard, lag=2 = +1 delay."""
|
||||
c = df["close"].values.astype(float)
|
||||
tgt, r = target_series(c, bpd)
|
||||
pos = np.zeros(len(tgt))
|
||||
pos[lag:] = tgt[:-lag]
|
||||
gross = pos * r
|
||||
turn = np.abs(np.diff(pos, prepend=0.0))
|
||||
net = gross - FEE_SIDE * turn
|
||||
net[:lag] = 0.0
|
||||
return np.clip(net, -0.99, None), pd.to_datetime(df["datetime"])
|
||||
|
||||
|
||||
def portfolio_metrics(dfs, bpd, lag):
|
||||
series = {}
|
||||
for a in ASSETS:
|
||||
net, ts = sleeve_net(dfs[a], bpd, lag)
|
||||
series[a] = pd.Series(net, index=pd.to_datetime(ts.values))
|
||||
J = pd.concat(series, axis=1, join="inner").dropna()
|
||||
combo = 0.5 * J["BTC"].values + 0.5 * J["ETH"].values
|
||||
bpy = bpd * 365.25
|
||||
sh = float(np.mean(combo) / np.std(combo) * np.sqrt(bpy)) if np.std(combo) > 0 else 0.0
|
||||
eq = np.cumprod(1.0 + np.clip(combo, -0.99, None))
|
||||
dd = float(np.max((np.maximum.accumulate(eq) - eq) / np.maximum.accumulate(eq)))
|
||||
yrs = (J.index[-1] - J.index[0]).days / 365.25
|
||||
cagr = eq[-1] ** (1 / yrs) - 1
|
||||
return sh, dd, cagr
|
||||
|
||||
|
||||
def main():
|
||||
raw = {a: load(a, "1h") for a in ASSETS}
|
||||
print("=" * 96)
|
||||
print("# LOOK-AHEAD / EXECUTION-LAG AUDIT — trend portfolio (long-flat, tvol20, lev2), per timeframe")
|
||||
print("# lag1 = standard (decision held next bar). lag2 = +1 bar execution delay (conservative).")
|
||||
print("# left/right = resample label (open vs close). Big gap => labeling leak.")
|
||||
print("=" * 96)
|
||||
print(f" {'TF':<5s}{'Sh lag1(L)':>12s}{'Sh lag2(L)':>12s}{'Sh lag1(R)':>12s}"
|
||||
f"{'CAGR l1':>10s}{'CAGR l2':>10s}{'DD l1':>8s}{'lag-decay':>11s}")
|
||||
for tf, (rule, bpd) in TFS.items():
|
||||
dfsL = {a: resample(raw[a], rule, "left") for a in ASSETS}
|
||||
dfsR = {a: resample(raw[a], rule, "right") for a in ASSETS}
|
||||
sh1L, dd1, cagr1 = portfolio_metrics(dfsL, bpd, 1)
|
||||
sh2L, _, cagr2 = portfolio_metrics(dfsL, bpd, 2)
|
||||
sh1R, _, _ = portfolio_metrics(dfsR, bpd, 1)
|
||||
decay = (sh1L - sh2L) / sh1L * 100 if sh1L else 0.0
|
||||
flag = " <-- robust" if sh2L >= 0.9 * sh1L and abs(sh1L - sh1R) < 0.1 else ""
|
||||
print(f" {tf:<5s}{sh1L:>12.2f}{sh2L:>12.2f}{sh1R:>12.2f}"
|
||||
f"{cagr1*100:>+9.1f}%{cagr2*100:>+9.1f}%{dd1*100:>7.1f}%{decay:>+10.0f}%{flag}")
|
||||
print("\n Interpretation:")
|
||||
print(" - If Sh lag2 << Sh lag1 (big lag-decay), the edge needs to trade AT the close -> sub-TF")
|
||||
print(" timing artifact / cost-fragile. Robust slow-trend should barely move with +1 bar.")
|
||||
print(" - If Sh lag1(left) != Sh lag1(right), the bar LABELING leaks -> look-ahead. Should match.")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,365 @@
|
||||
"""TRACK F — CALENDAR SEASONALITY on BTC & ETH (hour-of-day, day-of-week, interactions).
|
||||
|
||||
Honest test of whether there is a SYSTEMATIC, TRADEABLE calendar edge on the certified
|
||||
Deribit-mainnet BTC/ETH feeds. Seasonality is the easiest place on earth to overfit
|
||||
(24 hours x 7 weekdays = 168 buckets => you WILL find "significant" cells by chance), so
|
||||
every claim here is held to the project's anti-look-ahead, OOS, per-year, both-assets bar.
|
||||
|
||||
METHODOLOGY (no shortcuts):
|
||||
- ret[i] = close[i]/close[i-1]-1 is known at close[i]. A position decided at close[i]
|
||||
earns ret[i+1]. We NEVER include the bar being traded (or any future bar) in the
|
||||
statistic that decides the trade.
|
||||
- DESCRIPTIVE tables (per-hour / per-weekday mean returns) are split IS(65%)/OOS(35%).
|
||||
They are diagnostics, not trades.
|
||||
- TRADEABLE rule = ADAPTIVE EXPANDING sign: at close[i] we look up the calendar bucket
|
||||
of bar i+1 (the clock is known with zero look-ahead) and take the SIGN of that bucket's
|
||||
mean return computed ONLY on bars <= i (expanding, warmup-gated). Long-flat or
|
||||
long-short. Fees charged only on |Δposition| (turnover-aware). This lets the data pick
|
||||
each bucket's sign LIVE — the honest analogue of "trade the seasonal".
|
||||
- Also an in-sample-optimised discrete rule (enter at hour H, hold W bars, best dir) is
|
||||
shown ONLY to demonstrate the overfit gap IS->OOS.
|
||||
- NET fees fee_side baseline 0.0005 (=0.10% RT); swept 0.0005/0.00075/0.001.
|
||||
- A survivor must be net-positive OOS AND across years AND on BOTH BTC & ETH.
|
||||
|
||||
Run: uv run python scripts/research/trackF_seasonality.py
|
||||
"""
|
||||
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 # noqa: E402
|
||||
|
||||
ASSETS = ["BTC", "ETH"]
|
||||
TF = "1h"
|
||||
FEE_SIDE = 0.0005 # 0.05%/side = 0.10% round-trip
|
||||
BARS_PER_DAY = 24
|
||||
BPY = BARS_PER_DAY * 365.25
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
def prep(asset: str, tf: str = TF):
|
||||
df = load(asset, tf)
|
||||
c = df["close"].values.astype(float)
|
||||
ret = np.empty(len(c))
|
||||
ret[0] = 0.0
|
||||
ret[1:] = c[1:] / c[:-1] - 1.0
|
||||
dt = pd.to_datetime(df["datetime"])
|
||||
return dict(
|
||||
df=df, ret=ret,
|
||||
hour=dt.dt.hour.values.astype(int),
|
||||
dow=dt.dt.dayofweek.values.astype(int), # 0=Mon..6=Sun
|
||||
ts=dt,
|
||||
)
|
||||
|
||||
|
||||
def metrics_from_pnl(pnl: np.ndarray, ts: pd.Series):
|
||||
"""pnl[i] = realized per-bar net return of the strategy (already fee-adjusted)."""
|
||||
eq = np.cumprod(1.0 + np.clip(pnl, -0.99, None))
|
||||
r = pnl[np.isfinite(pnl)]
|
||||
sharpe = float(np.mean(r) / np.std(r) * np.sqrt(BPY)) if np.std(r) > 0 else 0.0
|
||||
peak = np.maximum.accumulate(eq)
|
||||
maxdd = float(np.max((peak - eq) / peak)) if len(eq) else 0.0
|
||||
span_days = (ts.iloc[-1] - ts.iloc[0]).total_seconds() / 86400
|
||||
years = span_days / 365.25 if span_days > 0 else 1.0
|
||||
total = eq[-1] / eq[0] if len(eq) else 1.0
|
||||
cagr = total ** (1 / years) - 1 if years > 0 and total > 0 else -1.0
|
||||
daily_2k = (2000 * total - 2000) / span_days if span_days > 0 else 0.0
|
||||
return dict(sharpe=sharpe, maxdd=maxdd, cagr=cagr, total=total - 1.0,
|
||||
daily_2k=daily_2k, eq=eq)
|
||||
|
||||
|
||||
def per_year_pnl(pnl: np.ndarray, ts: pd.Series):
|
||||
s = pd.Series(pnl, index=ts.values)
|
||||
out = {}
|
||||
for y, g in s.groupby(s.index.year):
|
||||
eq = np.cumprod(1.0 + np.clip(g.values, -0.99, None))
|
||||
out[int(y)] = float(eq[-1] - 1.0)
|
||||
return out
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 1. DESCRIPTIVE seasonality tables (diagnostics, IS vs OOS)
|
||||
# ---------------------------------------------------------------------------
|
||||
def descriptive(data, frac=0.65):
|
||||
n = len(data["ret"])
|
||||
cut = int(n * frac)
|
||||
ret, hour, dow = data["ret"], data["hour"], data["dow"]
|
||||
rows_h, rows_d = {}, {}
|
||||
for h in range(24):
|
||||
m_is = ret[:cut][hour[:cut] == h]
|
||||
m_oos = ret[cut:][hour[cut:] == h]
|
||||
rows_h[h] = (m_is.mean() * 1e4, m_oos.mean() * 1e4,
|
||||
np.sign(m_is.mean()) == np.sign(m_oos.mean()))
|
||||
for d in range(7):
|
||||
m_is = ret[:cut][dow[:cut] == d]
|
||||
m_oos = ret[cut:][dow[cut:] == d]
|
||||
rows_d[d] = (m_is.mean() * 1e4, m_oos.mean() * 1e4,
|
||||
np.sign(m_is.mean()) == np.sign(m_oos.mean()))
|
||||
return rows_h, rows_d
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 2. ADAPTIVE EXPANDING-sign seasonal strategy (the honest tradeable test)
|
||||
# ---------------------------------------------------------------------------
|
||||
def adaptive_seasonal(data, bucket="hour", mode="longshort",
|
||||
warmup=200, fee_side=FEE_SIDE):
|
||||
"""Position at close[i] = sign of the EXPANDING past mean return of bar (i+1)'s
|
||||
calendar bucket, using only bars <= i. earns ret[i+1]. Fee on |Δposition|."""
|
||||
ret = data["ret"]
|
||||
key = data[bucket]
|
||||
n = len(ret)
|
||||
nbuck = int(key.max()) + 1
|
||||
sums = np.zeros(nbuck)
|
||||
counts = np.zeros(nbuck)
|
||||
pos = np.zeros(n)
|
||||
for i in range(1, n - 1):
|
||||
b = key[i]
|
||||
sums[b] += ret[i]
|
||||
counts[b] += 1
|
||||
nb = key[i + 1]
|
||||
if counts[nb] >= warmup:
|
||||
m = sums[nb] / counts[nb]
|
||||
if m > 0:
|
||||
pos[i] = 1.0
|
||||
else:
|
||||
pos[i] = -1.0 if mode == "longshort" else 0.0
|
||||
# pnl[i] earned over bar i+1
|
||||
pnl = np.zeros(n)
|
||||
prev = 0.0
|
||||
for i in range(1, n - 1):
|
||||
turn = abs(pos[i] - prev)
|
||||
pnl[i] = pos[i] * ret[i + 1] - fee_side * turn
|
||||
prev = pos[i]
|
||||
return pnl, pos
|
||||
|
||||
|
||||
def adaptive_hourxdow(data, mode="longshort", warmup=120, fee_side=FEE_SIDE):
|
||||
ret, hour, dow = data["ret"], data["hour"], data["dow"]
|
||||
key = hour * 7 + dow # 168 buckets
|
||||
n = len(ret)
|
||||
sums = np.zeros(168)
|
||||
counts = np.zeros(168)
|
||||
pos = np.zeros(n)
|
||||
for i in range(1, n - 1):
|
||||
b = key[i]
|
||||
sums[b] += ret[i]
|
||||
counts[b] += 1
|
||||
nb = key[i + 1]
|
||||
if counts[nb] >= warmup:
|
||||
m = sums[nb] / counts[nb]
|
||||
if m > 0:
|
||||
pos[i] = 1.0
|
||||
else:
|
||||
pos[i] = -1.0 if mode == "longshort" else 0.0
|
||||
pnl = np.zeros(n)
|
||||
prev = 0.0
|
||||
for i in range(1, n - 1):
|
||||
turn = abs(pos[i] - prev)
|
||||
pnl[i] = pos[i] * ret[i + 1] - fee_side * turn
|
||||
prev = pos[i]
|
||||
return pnl, pos
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 3. In-sample-optimised DISCRETE rule (to expose the overfit gap)
|
||||
# ---------------------------------------------------------------------------
|
||||
def discrete_hour_rule_scan(data, frac=0.65, fee_side=FEE_SIDE):
|
||||
"""Scan IS for best (entry_hour, hold_window, direction) by IS Sharpe; report OOS.
|
||||
|
||||
A trade: enter at close of bar whose hour==H (decided with data<=close[i]), hold W
|
||||
bars, exit at close. One trade per day. Fee charged round-trip on each trade.
|
||||
"""
|
||||
ret, hour, ts = data["ret"], data["hour"], data["ts"]
|
||||
n = len(ret)
|
||||
cut = int(n * frac)
|
||||
|
||||
def rule_pnl(H, W, direction, lo, hi):
|
||||
pnl = np.zeros(n)
|
||||
i = lo
|
||||
last_exit = lo - 1
|
||||
while i < hi:
|
||||
if hour[i] == H and i > last_exit:
|
||||
# cumulative return over the next W bars: prod(1+ret[i+1..i+W]) - 1
|
||||
end = min(i + W, n - 1)
|
||||
gross = np.prod(1.0 + ret[i + 1:end + 1]) - 1.0
|
||||
pnl[i] = direction * gross - 2 * fee_side
|
||||
last_exit = end
|
||||
i = end
|
||||
else:
|
||||
i += 1
|
||||
return pnl
|
||||
|
||||
best = None
|
||||
n_tested = 0
|
||||
for H in range(24):
|
||||
for W in (1, 2, 3, 4, 6, 8, 12, 24):
|
||||
for direction in (+1, -1):
|
||||
n_tested += 1
|
||||
pnl_is = rule_pnl(H, W, direction, 1, cut)
|
||||
r = pnl_is[pnl_is != 0.0]
|
||||
if len(r) < 50:
|
||||
continue
|
||||
sh = np.mean(r) / np.std(r) * np.sqrt(BPY) if np.std(r) > 0 else 0.0
|
||||
if best is None or sh > best[0]:
|
||||
best = (sh, H, W, direction)
|
||||
sh, H, W, direction = best
|
||||
pnl_oos = rule_pnl(H, W, direction, cut, n)
|
||||
r_oos = pnl_oos[pnl_oos != 0.0]
|
||||
sh_oos = (np.mean(r_oos) / np.std(r_oos) * np.sqrt(BPY)) if (len(r_oos) and np.std(r_oos) > 0) else 0.0
|
||||
return dict(n_tested=n_tested, H=H, W=W, dir=direction, sh_is=sh,
|
||||
sh_oos=sh_oos, n_is=int((rule_pnl(H, W, direction, 1, cut) != 0).sum()),
|
||||
n_oos=len(r_oos), oos_mean_bp=r_oos.mean() * 1e4 if len(r_oos) else 0.0)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# reporting
|
||||
# ---------------------------------------------------------------------------
|
||||
def split_metrics(pnl, ts, frac=0.65):
|
||||
n = len(pnl)
|
||||
cut = int(n * frac)
|
||||
m_is = metrics_from_pnl(pnl[:cut], ts.iloc[:cut])
|
||||
m_oos = metrics_from_pnl(pnl[cut:], ts.iloc[cut:])
|
||||
m_all = metrics_from_pnl(pnl, ts)
|
||||
return m_is, m_oos, m_all
|
||||
|
||||
|
||||
def turnover_per_year(pos, ts):
|
||||
s = pd.Series(np.abs(np.diff(pos, prepend=0.0)), index=ts.values)
|
||||
return s.groupby(s.index.year).sum().to_dict()
|
||||
|
||||
|
||||
def main():
|
||||
print("=" * 100)
|
||||
print("# TRACK F — CALENDAR SEASONALITY (hour-of-day / day-of-week / hour×weekday)")
|
||||
print("# certified Deribit-mainnet BTC & ETH, 1h UTC. fee_side=0.0005 (0.10% RT).")
|
||||
print("# No look-ahead: bucket stats use only bars <= i; position earns ret[i+1].")
|
||||
print("=" * 100)
|
||||
|
||||
data = {a: prep(a) for a in ASSETS}
|
||||
|
||||
# --- DESCRIPTIVE ---------------------------------------------------------
|
||||
print("\n" + "#" * 100)
|
||||
print("# 1. DESCRIPTIVE per-bucket mean returns (basis points/bar). IS=first 65%, OOS=last 35%.")
|
||||
print("# 'sign?' = IS and OOS agree on sign. Diagnostics only (NOT trades, no fees).")
|
||||
print("#" * 100)
|
||||
for a in ASSETS:
|
||||
rows_h, rows_d = descriptive(data[a])
|
||||
print(f"\n ── {a} HOUR-OF-DAY (UTC) mean bp/hr ─────────────────────────────")
|
||||
print(" hr : IS_bp OOS_bp sign?")
|
||||
agree_h = 0
|
||||
for h in range(24):
|
||||
iv, ov, ag = rows_h[h]
|
||||
agree_h += int(ag)
|
||||
flag = " <-- US open" if h in (13, 14) else (" <-- US close" if h in (20, 21) else "")
|
||||
print(f" {h:>2d} : {iv:>+6.2f} {ov:>+6.2f} {'Y' if ag else '.'}{flag}")
|
||||
print(f" hour sign-agreement IS/OOS: {agree_h}/24")
|
||||
print(f"\n ── {a} DAY-OF-WEEK mean bp/bar (0=Mon..6=Sun) ──────────────────")
|
||||
names = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"]
|
||||
agree_d = 0
|
||||
for d in range(7):
|
||||
iv, ov, ag = rows_d[d]
|
||||
agree_d += int(ag)
|
||||
print(f" {names[d]} : {iv:>+6.3f} {ov:>+6.3f} {'Y' if ag else '.'}")
|
||||
print(f" weekday sign-agreement IS/OOS: {agree_d}/7")
|
||||
|
||||
# --- ADAPTIVE EXPANDING-SIGN (the honest tradeable test) ----------------
|
||||
print("\n" + "#" * 100)
|
||||
print("# 2. ADAPTIVE EXPANDING-SIGN seasonal strategies (HONEST tradeable test).")
|
||||
print("# sign of bucket's PAST-ONLY mean decides position; fee on turnover.")
|
||||
print("#" * 100)
|
||||
configs = [
|
||||
("HOUR long-short", "hour", "longshort", 200),
|
||||
("HOUR long-flat ", "hour", "longflat", 200),
|
||||
("DOW long-short", "dow", "longshort", 60),
|
||||
("DOW long-flat ", "dow", "longflat", 60),
|
||||
]
|
||||
for label, bucket, mode, warmup in configs:
|
||||
print(f"\n ── {label} ────────────────────────────────────────────────────")
|
||||
for a in ASSETS:
|
||||
pnl, pos = adaptive_seasonal(data[a], bucket=bucket, mode=mode, warmup=warmup)
|
||||
ts = data[a]["ts"]
|
||||
m_is, m_oos, m_all = split_metrics(pnl, ts)
|
||||
py = per_year_pnl(pnl, ts)
|
||||
yrs = "".join(f"{py.get(y, float('nan'))*100:>+6.0f}" for y in range(2019, 2027))
|
||||
print(f" {a}: ALL Sh={m_all['sharpe']:>+5.2f} CAGR={m_all['cagr']*100:>+6.1f}% "
|
||||
f"DD={m_all['maxdd']*100:>4.1f}% €/d={m_all['daily_2k']:>+5.2f} | "
|
||||
f"IS Sh={m_is['sharpe']:>+5.2f} OOS Sh={m_oos['sharpe']:>+5.2f}")
|
||||
print(f" per-year %: {yrs} (2019..2026)")
|
||||
|
||||
# buy-and-hold benchmark — the key control: does any 'seasonal' beat just being long?
|
||||
print(f"\n ── BUY-AND-HOLD benchmark (the control for long-bias) ──")
|
||||
for a in ASSETS:
|
||||
ret = data[a]["ret"].copy()
|
||||
ret[0] = 0.0
|
||||
m = metrics_from_pnl(ret, data[a]["ts"])
|
||||
print(f" {a}: Sh={m['sharpe']:>+5.2f} CAGR={m['cagr']*100:>+6.1f}% DD={m['maxdd']*100:>4.1f}% "
|
||||
f" <- compare to DOW long-flat above (it's nearly identical = no edge, just long)")
|
||||
|
||||
# hour x weekday interaction (168 buckets — extreme overfit risk)
|
||||
print(f"\n ── HOUR×WEEKDAY long-short (168 buckets, warmup 120) — overfit canary ──")
|
||||
for a in ASSETS:
|
||||
pnl, pos = adaptive_hourxdow(data[a], mode="longshort", warmup=120)
|
||||
ts = data[a]["ts"]
|
||||
m_is, m_oos, m_all = split_metrics(pnl, ts)
|
||||
print(f" {a}: ALL Sh={m_all['sharpe']:>+5.2f} CAGR={m_all['cagr']*100:>+6.1f}% "
|
||||
f"DD={m_all['maxdd']*100:>4.1f}% | IS Sh={m_is['sharpe']:>+5.2f} OOS Sh={m_oos['sharpe']:>+5.2f}")
|
||||
|
||||
# --- FEE SWEEP on the best adaptive config -------------------------------
|
||||
print("\n" + "#" * 100)
|
||||
print("# 3. FEE SWEEP — HOUR long-short adaptive (turnover-aware). Are survivors fee-robust?")
|
||||
print("#" * 100)
|
||||
for fee in (0.0, 0.0005, 0.00075, 0.001):
|
||||
line = f" fee_side={fee:.5f} (RT {fee*2*100:.2f}%): "
|
||||
for a in ASSETS:
|
||||
pnl, _ = adaptive_seasonal(data[a], bucket="hour", mode="longshort",
|
||||
warmup=200, fee_side=fee)
|
||||
m = metrics_from_pnl(pnl, data[a]["ts"])
|
||||
line += f"{a} Sh={m['sharpe']:>+5.2f} CAGR={m['cagr']*100:>+6.1f}% "
|
||||
print(line)
|
||||
|
||||
# --- TURNOVER (fees are first-order for hour strategies) -----------------
|
||||
print("\n" + "#" * 100)
|
||||
print("# 4. TURNOVER (HOUR long-short adaptive): position flips/year (each flip costs ~fee).")
|
||||
print("#" * 100)
|
||||
for a in ASSETS:
|
||||
_, pos = adaptive_seasonal(data[a], bucket="hour", mode="longshort", warmup=200)
|
||||
tpy = turnover_per_year(pos, data[a]["ts"])
|
||||
s = " ".join(f"{y}:{int(v)}" for y, v in sorted(tpy.items()))
|
||||
print(f" {a} turnover units/yr: {s}")
|
||||
|
||||
# --- IN-SAMPLE-OPTIMISED DISCRETE RULE (overfit demonstration) ----------
|
||||
print("\n" + "#" * 100)
|
||||
print("# 5. IN-SAMPLE-OPTIMISED discrete rule (enter hour H, hold W, best dir).")
|
||||
print("# Picked by IS Sharpe, reported OOS. Demonstrates the multiple-testing trap.")
|
||||
print("#" * 100)
|
||||
for a in ASSETS:
|
||||
r = discrete_hour_rule_scan(data[a])
|
||||
print(f" {a}: tested {r['n_tested']} (H,W,dir) cells -> best IS "
|
||||
f"H={r['H']:02d} hold={r['W']}h dir={r['dir']:+d} "
|
||||
f"IS Sh={r['sh_is']:>+5.2f} (n={r['n_is']}) -> OOS Sh={r['sh_oos']:>+5.2f} "
|
||||
f"(n={r['n_oos']}, mean {r['oos_mean_bp']:>+.1f} bp/trade)")
|
||||
|
||||
# --- VERDICT -------------------------------------------------------------
|
||||
print("\n" + "#" * 100)
|
||||
print("# MULTIPLE-TESTING CAVEAT")
|
||||
print("#" * 100)
|
||||
print("""
|
||||
Buckets examined: 24 hours + 7 weekdays + 168 hour×weekday = 199 calendar cells PER ASSET,
|
||||
each tested IS and OOS, plus discrete grid = 24×8×2 = 384 (H,W,dir) cells per asset.
|
||||
With that many cells, spurious 'significant' buckets are GUARANTEED. The honest filters
|
||||
applied here: (a) adaptive sign chosen live on PAST data only (no cherry-picking),
|
||||
(b) must hold OOS, (c) must hold per-year, (d) must hold on BOTH BTC AND ETH.
|
||||
Read the IS->OOS Sharpe collapse and the per-year sign flips above as the real verdict.
|
||||
""")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,478 @@
|
||||
"""TRACK G — PRIOR-PERIOD LEVEL BREAKOUTS / RANGE on CLEAN BTC/ETH (Deribit mainnet).
|
||||
|
||||
HONEST harness only. We test rules defined RELATIVE TO A PRIOR CALENDAR PERIOD:
|
||||
* prior-DAY high/low breakout (continuation AND fade)
|
||||
* opening-range breakout (first N UTC hours -> break for rest of day)
|
||||
* prior-day CLOSE / gap / range-position / prior-day return-sign filter
|
||||
* prior-WEEK high/low breakout
|
||||
* time-anchored entries (act at a given UTC hour vs prior-day level), exit EOD/fixed/TP-SL
|
||||
|
||||
The single question: on clean BTC/ETH, with a genuinely EXECUTABLE entry (direction and
|
||||
price decided with data <= close[i], fill at close[i], NEVER entering at the exact level
|
||||
intrabar), net of realistic Deribit fees, OOS and grid-robust on BOTH assets —
|
||||
do prior-period breakouts CONTINUE (trend) or REVERT (fade)? Is there a deployable edge?
|
||||
|
||||
NO LOOK-AHEAD GUARANTEES:
|
||||
* Prior-period levels are built by aggregating to daily/weekly bars and SHIFTING by one
|
||||
full period (shift(1) on the closed-period frame). 'Today'/'this-week' is NEVER part of
|
||||
the level. The prior period is fully closed before any bar of the current period.
|
||||
* Opening-range levels are used ONLY on bars AFTER the open window has fully closed.
|
||||
* Direction + price decided at close[i]; fill at close[i] (harness enforces).
|
||||
|
||||
Run:
|
||||
uv run python scripts/research/trackG_prior_levels.py # full
|
||||
uv run python scripts/research/trackG_prior_levels.py --quick # 1h only, fewer grids
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
import time
|
||||
from itertools import product
|
||||
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
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Causal helpers
|
||||
# ===========================================================================
|
||||
def atr(df: pd.DataFrame, period: int = 14) -> 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
|
||||
|
||||
|
||||
def prior_period_levels(df: pd.DataFrame, period: str = "D") -> dict:
|
||||
"""Return prior-period high/low/close/open/range arrays aligned to each intraday bar.
|
||||
|
||||
period='D': prior calendar day (UTC). period='W': prior ISO week (anchored Mon 00:00 UTC).
|
||||
Uses shift(1) on the CLOSED-period frame: the level for the current period only sees the
|
||||
fully-closed previous period -> no look-ahead.
|
||||
"""
|
||||
dt = df["datetime"]
|
||||
if period == "D":
|
||||
key = dt.dt.floor("D")
|
||||
elif period == "W":
|
||||
key = dt.dt.floor("D") - pd.to_timedelta(dt.dt.weekday, unit="D")
|
||||
else:
|
||||
raise ValueError(period)
|
||||
|
||||
key = key.reset_index(drop=True)
|
||||
agg = pd.DataFrame({
|
||||
"key": key,
|
||||
"high": df["high"].values, "low": df["low"].values,
|
||||
"close": df["close"].values, "open": df["open"].values,
|
||||
})
|
||||
g = agg.groupby("key").agg(high=("high", "max"), low=("low", "min"),
|
||||
close=("close", "last"), open=("open", "first")).sort_index()
|
||||
gp = g.shift(1) # prior, fully-closed period
|
||||
km = key.map # map current-period key -> prior-period aggregate
|
||||
ph = km(gp["high"]).values.astype(float)
|
||||
pl = km(gp["low"]).values.astype(float)
|
||||
pc = km(gp["close"]).values.astype(float)
|
||||
po = km(gp["open"]).values.astype(float)
|
||||
pret = (gp["close"] / gp["open"] - 1.0) # prior-period return (sign filter)
|
||||
prv = key.map(pret).values.astype(float)
|
||||
return {"ph": ph, "pl": pl, "pc": pc, "po": po, "prange": ph - pl, "pret": prv}
|
||||
|
||||
|
||||
def opening_range(df: pd.DataFrame, n_open_hours: int) -> dict:
|
||||
"""Opening-range high/low for the first n_open_hours of each UTC day, plus a per-bar
|
||||
flag of whether the open window has CLOSED (hour >= n_open_hours)."""
|
||||
dt = df["datetime"]
|
||||
date = dt.dt.floor("D")
|
||||
hour = dt.dt.hour
|
||||
date = date.reset_index(drop=True)
|
||||
in_open = (hour < n_open_hours).values
|
||||
o = pd.DataFrame({"date": date, "high": df["high"].values, "low": df["low"].values})
|
||||
o_open = o[in_open]
|
||||
org = o_open.groupby("date").agg(orh=("high", "max"), orl=("low", "min"))
|
||||
orh = date.map(org["orh"]).values.astype(float)
|
||||
orl = date.map(org["orl"]).values.astype(float)
|
||||
closed = (hour >= n_open_hours).values
|
||||
return {"orh": orh, "orl": orl, "closed": closed}
|
||||
|
||||
|
||||
def bars_left_in_day(df: pd.DataFrame) -> np.ndarray:
|
||||
date = df["datetime"].dt.floor("D")
|
||||
grp = df.groupby(date)
|
||||
idx_in_day = grp.cumcount().values
|
||||
size = grp["close"].transform("size").values
|
||||
return (size - idx_in_day - 1).astype(int)
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Signal generators -> list[dict|None] length len(df). Decisions use data <= close[i].
|
||||
# ===========================================================================
|
||||
def sig_prior_break(df, period="D", level="high", side="cont", anchor_hour=None,
|
||||
exit_mode="eod", max_bars=24, tp_atr=0.0, sl_atr=0.0, atr_p=14,
|
||||
buffer=0.0):
|
||||
"""Prior-period level breakout.
|
||||
level='high': trigger when close[i] > prior_high*(1+buffer)
|
||||
level='low' : trigger when close[i] < prior_low *(1-buffer)
|
||||
side='cont' : trade IN the breakout direction (high->long, low->short)
|
||||
side='fade' : trade AGAINST it (high->short, low->long)
|
||||
anchor_hour : if set, only evaluate on bars at that UTC hour (time-anchored)
|
||||
exit_mode : 'eod' (close at end of UTC day), 'bars' (max_bars), TP/SL via *_atr.
|
||||
"""
|
||||
lv = prior_period_levels(df, period)
|
||||
c = df["close"].values
|
||||
a = atr(df, atr_p) if (tp_atr or sl_atr) else None
|
||||
bl = bars_left_in_day(df) if exit_mode == "eod" else None
|
||||
hour = df["datetime"].dt.hour.values
|
||||
n = len(c)
|
||||
out = [None] * n
|
||||
ref = lv["ph"] if level == "high" else lv["pl"]
|
||||
for i in range(n):
|
||||
if anchor_hour is not None and hour[i] != anchor_hour:
|
||||
continue
|
||||
r = ref[i]
|
||||
if not np.isfinite(r):
|
||||
continue
|
||||
px = c[i]
|
||||
if level == "high":
|
||||
if not (px > r * (1.0 + buffer)):
|
||||
continue
|
||||
brk_dir = 1
|
||||
else:
|
||||
if not (px < r * (1.0 - buffer)):
|
||||
continue
|
||||
brk_dir = -1
|
||||
direction = brk_dir if side == "cont" else -brk_dir
|
||||
if exit_mode == "eod":
|
||||
mb = max(int(bl[i]), 1)
|
||||
else:
|
||||
mb = max_bars
|
||||
tp = sl = None
|
||||
if a is not None and np.isfinite(a[i]):
|
||||
if tp_atr:
|
||||
tp = px + direction * tp_atr * a[i]
|
||||
if sl_atr:
|
||||
sl = px - direction * sl_atr * a[i]
|
||||
out[i] = {"dir": direction, "tp": tp, "sl": sl, "max_bars": mb}
|
||||
return out
|
||||
|
||||
|
||||
def sig_or_break(df, n_open_hours=6, side="cont", exit_mode="eod", max_bars=12,
|
||||
tp_atr=0.0, sl_atr=0.0, atr_p=14, buffer=0.0):
|
||||
"""Opening-range breakout: after the first n_open_hours close, trade a break of the
|
||||
OR high (long if cont) or OR low (short if cont). Only the FIRST break per day fires
|
||||
(the harness keeps the position busy until exit)."""
|
||||
orr = opening_range(df, n_open_hours)
|
||||
c = df["close"].values
|
||||
a = atr(df, atr_p) if (tp_atr or sl_atr) else None
|
||||
bl = bars_left_in_day(df) if exit_mode == "eod" else None
|
||||
n = len(c)
|
||||
out = [None] * n
|
||||
orh, orl, closed = orr["orh"], orr["orl"], orr["closed"]
|
||||
for i in range(n):
|
||||
if not closed[i] or not np.isfinite(orh[i]):
|
||||
continue
|
||||
px = c[i]
|
||||
if px > orh[i]:
|
||||
brk = 1
|
||||
elif px < orl[i]:
|
||||
brk = -1
|
||||
else:
|
||||
continue
|
||||
direction = brk if side == "cont" else -brk
|
||||
if exit_mode == "eod":
|
||||
mb = max(int(bl[i]), 1)
|
||||
else:
|
||||
mb = max_bars
|
||||
tp = sl = None
|
||||
if a is not None and np.isfinite(a[i]):
|
||||
if tp_atr:
|
||||
tp = px + direction * tp_atr * a[i]
|
||||
if sl_atr:
|
||||
sl = px - direction * sl_atr * a[i]
|
||||
out[i] = {"dir": direction, "tp": tp, "sl": sl, "max_bars": mb}
|
||||
return out
|
||||
|
||||
|
||||
def sig_gap(df, side="cont", anchor_hour=0, thr=0.0, exit_mode="eod", max_bars=24,
|
||||
ret_filter=0):
|
||||
"""Gap vs prior-day CLOSE, evaluated at a given UTC hour (default the first bar of the
|
||||
day). gap = close[i]/prior_close - 1. If gap>thr -> up-gap; gap<-thr -> down-gap.
|
||||
side='cont' trades in the gap direction; 'fade' against. ret_filter: +1 only when
|
||||
prior-day return positive, -1 only when negative, 0 no filter."""
|
||||
lv = prior_period_levels(df, "D")
|
||||
c = df["close"].values
|
||||
bl = bars_left_in_day(df) if exit_mode == "eod" else None
|
||||
hour = df["datetime"].dt.hour.values
|
||||
pc, pret = lv["pc"], lv["pret"]
|
||||
n = len(c)
|
||||
out = [None] * n
|
||||
for i in range(n):
|
||||
if hour[i] != anchor_hour or not np.isfinite(pc[i]):
|
||||
continue
|
||||
gap = c[i] / pc[i] - 1.0
|
||||
if gap > thr:
|
||||
g = 1
|
||||
elif gap < -thr:
|
||||
g = -1
|
||||
else:
|
||||
continue
|
||||
if ret_filter and np.isfinite(pret[i]):
|
||||
if ret_filter > 0 and not (pret[i] > 0):
|
||||
continue
|
||||
if ret_filter < 0 and not (pret[i] < 0):
|
||||
continue
|
||||
direction = g if side == "cont" else -g
|
||||
mb = max(int(bl[i]), 1) if exit_mode == "eod" else max_bars
|
||||
out[i] = {"dir": direction, "tp": None, "sl": None, "max_bars": mb}
|
||||
return out
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Evaluation
|
||||
# ===========================================================================
|
||||
def run_split(df, sigfn, params, fee_rt=0.001, leverage=1.0, frac=0.65):
|
||||
cut = oos_split(df, frac)
|
||||
full = backtest_signals(df, sigfn(df, **params), fee_rt=fee_rt, leverage=leverage)
|
||||
di = df.iloc[:cut].reset_index(drop=True)
|
||||
do = df.iloc[cut:].reset_index(drop=True)
|
||||
is_ = backtest_signals(di, sigfn(di, **params), fee_rt=fee_rt, leverage=leverage)
|
||||
oos = backtest_signals(do, sigfn(do, **params), fee_rt=fee_rt, leverage=leverage)
|
||||
return full, is_, oos
|
||||
|
||||
|
||||
def hdr(t):
|
||||
print("\n" + "=" * 100)
|
||||
print(t)
|
||||
print("=" * 100)
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Main
|
||||
# ===========================================================================
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--quick", action="store_true")
|
||||
args = ap.parse_args()
|
||||
t0 = time.time()
|
||||
assets = ["BTC", "ETH"]
|
||||
tfs = ["1h"] if args.quick else ["1h", "15m"]
|
||||
|
||||
data = {}
|
||||
hdr("DATA")
|
||||
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()}")
|
||||
|
||||
# ---------------------------------------------------------------------
|
||||
# PASS 1 — PRIOR-DAY BREAKOUT: continuation vs fade, any-bar (first break/day),
|
||||
# EOD exit. THE core question: do prior-day breakouts continue or revert?
|
||||
# ---------------------------------------------------------------------
|
||||
hdr("PASS 1 — PRIOR-DAY HIGH/LOW breakout, any-bar first-break, EOD exit (1h, fee=0.001)\n"
|
||||
" CONTINUATION vs FADE side-by-side. OOS net must be >0 on BOTH to matter.")
|
||||
print(f" {'rule':<26s} | "
|
||||
f"{'BTC IS / OOS (tr, wr, shrp)':<40s} | {'ETH IS / OOS (tr, wr, shrp)':<40s}")
|
||||
for level in ["high", "low"]:
|
||||
for side in ["cont", "fade"]:
|
||||
name = f"PD {level:<4s} {side}"
|
||||
line = f" {name:<26s} | "
|
||||
for a in assets:
|
||||
df = data[(a, "1h")]
|
||||
_, is_, oos = run_split(df, sig_prior_break,
|
||||
dict(period="D", level=level, side=side,
|
||||
exit_mode="eod"))
|
||||
line += (f"{is_.net_return*100:>+6.0f}/{oos.net_return*100:>+6.0f}% "
|
||||
f"(t{oos.n_trades:>4d} w{oos.win_rate:>4.1f} s{oos.sharpe:>+4.1f}) | ")
|
||||
print(line)
|
||||
|
||||
# ---------------------------------------------------------------------
|
||||
# PASS 2 — OPENING-RANGE breakout (continuation vs fade), various open windows.
|
||||
# ---------------------------------------------------------------------
|
||||
hdr("PASS 2 — OPENING-RANGE breakout (first N UTC hours), EOD exit (1h, fee=0.001).\n"
|
||||
" CONTINUATION vs FADE. Survivor = OOS>0 on BOTH assets.")
|
||||
for nopen in ([6] if args.quick else [3, 6, 8, 12]):
|
||||
for side in ["cont", "fade"]:
|
||||
name = f"OR N={nopen:<2d} {side}"
|
||||
line = f" {name:<26s} | "
|
||||
for a in assets:
|
||||
df = data[(a, "1h")]
|
||||
_, is_, oos = run_split(df, sig_or_break,
|
||||
dict(n_open_hours=nopen, side=side, exit_mode="eod"))
|
||||
line += (f"{a} OOS={oos.net_return*100:>+6.0f}% "
|
||||
f"(t{oos.n_trades:>4d} w{oos.win_rate:>4.1f} s{oos.sharpe:>+4.1f}) | ")
|
||||
print(line)
|
||||
|
||||
# ---------------------------------------------------------------------
|
||||
# PASS 3 — GAP vs prior close at day open (hour 0), continuation vs fade,
|
||||
# with optional prior-day return-sign filter.
|
||||
# ---------------------------------------------------------------------
|
||||
hdr("PASS 3 — GAP vs prior-day CLOSE at hour 0, EOD exit (1h, fee=0.001).\n"
|
||||
" continuation vs fade; thr = min |gap|.")
|
||||
for thr in ([0.0] if args.quick else [0.0, 0.005, 0.01]):
|
||||
for side in ["cont", "fade"]:
|
||||
name = f"GAP thr={thr*100:.1f}% {side}"
|
||||
line = f" {name:<26s} | "
|
||||
for a in assets:
|
||||
df = data[(a, "1h")]
|
||||
_, is_, oos = run_split(df, sig_gap,
|
||||
dict(side=side, anchor_hour=0, thr=thr, exit_mode="eod"))
|
||||
line += (f"{a} OOS={oos.net_return*100:>+6.0f}% "
|
||||
f"(t{oos.n_trades:>4d} w{oos.win_rate:>4.1f} s{oos.sharpe:>+4.1f}) | ")
|
||||
print(line)
|
||||
|
||||
# ---------------------------------------------------------------------
|
||||
# PASS 4 — PRIOR-WEEK high/low breakout (continuation vs fade), EOD exit.
|
||||
# ---------------------------------------------------------------------
|
||||
hdr("PASS 4 — PRIOR-WEEK HIGH/LOW breakout, any-bar first-break, EOD exit (1h, fee=0.001).")
|
||||
for level in ["high", "low"]:
|
||||
for side in ["cont", "fade"]:
|
||||
name = f"PW {level:<4s} {side}"
|
||||
line = f" {name:<26s} | "
|
||||
for a in assets:
|
||||
df = data[(a, "1h")]
|
||||
_, is_, oos = run_split(df, sig_prior_break,
|
||||
dict(period="W", level=level, side=side,
|
||||
exit_mode="eod"))
|
||||
line += (f"{a} IS={is_.net_return*100:>+6.0f}% OOS={oos.net_return*100:>+6.0f}% "
|
||||
f"(t{oos.n_trades:>4d} s{oos.sharpe:>+4.1f}) | ")
|
||||
print(line)
|
||||
|
||||
# ---------------------------------------------------------------------
|
||||
# PASS 5 — TIME-ANCHORED prior-day breakout: sweep the anchor hour to expose
|
||||
# whether any apparent edge is just a lucky single hour.
|
||||
# ---------------------------------------------------------------------
|
||||
hdr("PASS 5 — TIME-ANCHORED PD-high CONTINUATION across UTC anchor hours (1h, EOD exit).\n"
|
||||
" A real edge is NOT a single lucky hour. (full-sample net per hour.)")
|
||||
hours = list(range(0, 24, 1 if not args.quick else 3))
|
||||
for a in assets:
|
||||
df = data[(a, "1h")]
|
||||
cells = []
|
||||
for hh in hours:
|
||||
full, _, _ = run_split(df, sig_prior_break,
|
||||
dict(period="D", level="high", side="cont",
|
||||
anchor_hour=hh, exit_mode="eod"))
|
||||
cells.append((hh, full.net_return * 100, full.sharpe, full.n_trades))
|
||||
pos = sum(1 for _, r, _, _ in cells if r > 0)
|
||||
print(f" {a}: {pos}/{len(cells)} anchor-hours net>0 (full). "
|
||||
f"best={max(cells, key=lambda x: x[1])[0]}h "
|
||||
f"({max(c[1] for c in cells):+.0f}%) worst={min(c[1] for c in cells):+.0f}%")
|
||||
line = " " + " ".join(f"{hh:02d}h:{r:>+5.0f}" for hh, r, _, _ in cells)
|
||||
print(line)
|
||||
|
||||
# ---------------------------------------------------------------------
|
||||
# PASS 6 — GRID ROBUSTNESS on the best family from PASS 1-4. We grid the
|
||||
# PD-low CONTINUATION and FADE plus OR breakout, require OOS>0 on BOTH assets.
|
||||
# ---------------------------------------------------------------------
|
||||
hdr("PASS 6 — GRID ROBUSTNESS. Cell SURVIVES only if OOS net>0 on BOTH BTC AND ETH.")
|
||||
|
||||
def grid(label, fn, base, sweep, tf="1h", fee=0.001):
|
||||
keys = list(sweep.keys())
|
||||
rows, surv = [], []
|
||||
for combo in product(*[sweep[k] for k in keys]):
|
||||
params = dict(base); params.update(dict(zip(keys, combo)))
|
||||
res = {}
|
||||
for a in assets:
|
||||
_, is_, oos = run_split(data[(a, tf)], fn, params, fee_rt=fee)
|
||||
res[a] = oos
|
||||
ok = all(res[a].net_return > 0 for a in assets)
|
||||
rows.append((params, res, ok))
|
||||
if ok:
|
||||
surv.append((params, res))
|
||||
print(f" [{label}] {len(surv)}/{len(rows)} cells OOS>0 on BOTH assets")
|
||||
rows.sort(key=lambda r: np.mean([r[1][a].net_return for a in assets]), reverse=True)
|
||||
for params, res, ok in rows[:5]:
|
||||
tag = "OK " if ok else " -"
|
||||
pp = {k: params[k] for k in sweep}
|
||||
s = f" {tag}{pp} | "
|
||||
for a in assets:
|
||||
s += f"{a} OOS={res[a].net_return*100:>+6.0f}% (s{res[a].sharpe:>+4.1f}) "
|
||||
print(s)
|
||||
return surv
|
||||
|
||||
sweeps = []
|
||||
sweeps.append(grid("PD-low cont", sig_prior_break,
|
||||
dict(period="D", level="low", side="cont", exit_mode="eod"),
|
||||
dict(buffer=[0.0, 0.001, 0.003], anchor_hour=[None])))
|
||||
sweeps.append(grid("PD-low fade", sig_prior_break,
|
||||
dict(period="D", level="low", side="fade", exit_mode="eod"),
|
||||
dict(buffer=[0.0, 0.001, 0.003], anchor_hour=[None])))
|
||||
sweeps.append(grid("PD-high cont", sig_prior_break,
|
||||
dict(period="D", level="high", side="cont", exit_mode="eod"),
|
||||
dict(buffer=[0.0, 0.001, 0.003], anchor_hour=[None])))
|
||||
sweeps.append(grid("PD-high fade", sig_prior_break,
|
||||
dict(period="D", level="high", side="fade", exit_mode="eod"),
|
||||
dict(buffer=[0.0, 0.001, 0.003], anchor_hour=[None])))
|
||||
if not args.quick:
|
||||
sweeps.append(grid("OR cont", sig_or_break,
|
||||
dict(side="cont", exit_mode="eod"),
|
||||
dict(n_open_hours=[3, 6, 8, 12])))
|
||||
sweeps.append(grid("OR fade", sig_or_break,
|
||||
dict(side="fade", exit_mode="eod"),
|
||||
dict(n_open_hours=[3, 6, 8, 12])))
|
||||
|
||||
# ---------------------------------------------------------------------
|
||||
# PASS 7 — FEE SWEEP + per-year on the single best surviving rule (if any),
|
||||
# else on the least-bad PD rule, to show fee sensitivity and year stability.
|
||||
# ---------------------------------------------------------------------
|
||||
hdr("PASS 7 — FEE SWEEP + PER-YEAR on the best PD rule. fee=0 is GROSS (is the SIGN of\n"
|
||||
" the edge even right before fees?).")
|
||||
# pick best rule: scan the 4 PD sides at default, mean OOS over assets
|
||||
candidates = [
|
||||
("PD low cont", dict(period="D", level="low", side="cont", exit_mode="eod")),
|
||||
("PD low fade", dict(period="D", level="low", side="fade", exit_mode="eod")),
|
||||
("PD high cont", dict(period="D", level="high", side="cont", exit_mode="eod")),
|
||||
("PD high fade", dict(period="D", level="high", side="fade", exit_mode="eod")),
|
||||
]
|
||||
scored = []
|
||||
for nm, p in candidates:
|
||||
m = np.mean([run_split(data[(a, "1h")], sig_prior_break, p)[2].net_return for a in assets])
|
||||
scored.append((m, nm, p))
|
||||
scored.sort(reverse=True)
|
||||
best_nm, best_p = scored[0][1], scored[0][2]
|
||||
print(f" best-by-meanOOS PD rule: {best_nm} (meanOOS={scored[0][0]*100:+.0f}%)")
|
||||
fees = [0.0, 0.0005, 0.001, 0.0015, 0.002]
|
||||
for a in assets:
|
||||
df = data[(a, "1h")]
|
||||
line = f" {a} fee-sweep (RT%): "
|
||||
for f in fees:
|
||||
full, _, oos = run_split(df, sig_prior_break, best_p, fee_rt=f)
|
||||
line += f"{f*100:.2f}%[full={full.net_return*100:>+5.0f}/OOS={oos.net_return*100:>+5.0f}] "
|
||||
print(line)
|
||||
print(" per-year (full sample, fee=0.001):")
|
||||
for a in assets:
|
||||
df = data[(a, "1h")]
|
||||
full, _, _ = run_split(df, sig_prior_break, best_p)
|
||||
yrs = " ".join(f"{y}:{full.yearly[y]*100:>+5.0f}%" for y in sorted(full.yearly))
|
||||
print(f" {a}: trades={full.n_trades} Sharpe={full.sharpe:+.2f} "
|
||||
f"maxDD={full.max_dd*100:.0f}% EUR/d(2k)={full.daily_profit(2000):+.2f}")
|
||||
print(f" {yrs}")
|
||||
|
||||
# ---------------------------------------------------------------------
|
||||
# VERDICT
|
||||
# ---------------------------------------------------------------------
|
||||
hdr("VERDICT")
|
||||
total_surv = sum(len(s) for s in sweeps)
|
||||
if total_surv == 0:
|
||||
print(" ZERO grid cells produced OOS net>0 on BOTH BTC and ETH at baseline fees.")
|
||||
print(" => No robust prior-period breakout/fade edge on clean BTC/ETH. The continuation-")
|
||||
print(" vs-fade tables above show which SIDE (if any) is even net-positive in-sample;")
|
||||
print(" consult PASS 1-5 for direction. Not deployable.")
|
||||
else:
|
||||
print(f" {total_surv} grid cell(s) survived OOS>0 on both assets. Inspect PASS 6/7 and")
|
||||
print(" stress with fee sweep + per-year before trusting. List of survivors:")
|
||||
for s in sweeps:
|
||||
for params, res in s:
|
||||
ms = np.mean([res[a].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
Block a user