"""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()