Adriano/PythagorasGoal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
5ac4e16af8deae3ea8e4eaa501445fddfabacc63
PythagorasGoal/scripts/research/trackG_prior_levels.py
T
Adriano Dal Pastro 87af03955c research: porta artefatti da strategy-research-calendar (tracks F-I + eval crypto_backtest + lead OPZIONI/VRP) 
Dal branch parallelo strategy-research-calendar (continuazione della linea TP01). Porta su main il
record di ricerca + la fondazione del lead opzioni (NIENTE blob dati, niente codice in conflitto):
- Tracks F/G/H/I (seasonality/calendar, prior-levels, volume-vol, momentum-reversal): tutti
  NEGATIVI/spurii -> confermano il soffitto Sharpe ~1.3 su BTC/ETH direzionale (calendar = buy&hold
  travestito; mean-reversion morta anche a fee 0). Diari + script.
- trackD_lookahead_audit.py: audit anti-look-ahead (stesso esito del nostro fix >=12h).
- eval-crypto-backtest-options.md: valutazione strategia esterna crypto_backtest. Cross-valida TP01
  (il loro sleeve spot 12h ~ TP01: due ricerche indipendenti, stessa conclusione). Identifica il
  LEAD: sleeve income OPZIONI (vendita put settimanali delta-0.28, VRP IV>RV), scorrelato ~0.22 al
  trend -> via per superare il soffitto ~1.3.
- options_real_quote_check.py + cerbero-bite-mainnet-verified.md: VERIFICATO su QUOTE REALI Deribit
  mainnet (cerbero-bite/MCP = mainnet, bit-identico a ccxt.deribit). Premio reale (BID, con skew) =
  1.29x il modellato -> il backtest SOTTOSTIMA il premio; il rischio vero e' la CODA (short-vol) +
  liquidita' di roll in stress, non la magnitudine.

NB: lo sleeve opzioni e' un LEAD, NON deployato: prezzato da modello (BS su DVOL) + 1 snapshot in
regime calmo. Serve validazione real-chain multi-regime + stress crash + paper su testnet prima di
aggiungerlo al portafoglio. Portafoglio attivo invariato: TP01 70% + XS01 30%.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-19 20:24:16 +00:00

479 lines
22 KiB
Python
Raw Blame History

"""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 View Git Blame Copy Permalink