Adriano/PythagorasGoal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
116abf520325cf50ca8319c76c5e63a87afef5ad
PythagorasGoal/scripts/research/r0702_ell_rangecycle.py
T
Adriano Dal Pastro 76120b59c2 research(wave-0702bis): ondata video-claims — Elliott 3/3 e Albimarini 2/2 scartati, mappa capital scaling 2-5k 
Sei agenti, nessun sopravvissuto:
- ELL-A range-cycle: rumore (0/24 Bonferroni; nessuna cella weekly regge
  a tutte le 7 ancore). Lezione pandas: resample("7D", origin) IGNORA
  origin -> usare "168h" per le bande d'ancora weekly.
- ELL-B Fibonacci: l'edge apparente e' la POSIZIONE dei livelli, non i
  numeri (null location-matched: pctl 0.39-0.68); confluenza FAIL 4/4.
- ELL-C canale: Donchian travestito (non batte il Donchian equivalente,
  DSR 0.685, IS 1.40 -> HOLD -0.87; target 1.618 = caso; anchor-luck 4h).
- ALB-A diagonale: il condor stessa-scadenza la batte a ogni f; senza
  gate IV-rank tutte le strutture perdono (3a conferma: l'alpha del VRP
  e' il gate); fee-negativa su Deribit a qualsiasi size; 2o caso
  "0-perdite = Sharpe implausibile" dopo CC01.
- ALB-B claims: 82%/PF 5.16/"420%" consistente con zero skill (P=20-45%,
  78.6% delle finestre 6-mesi lo produce); replay con code reali =
  rovina 1998/2002/2020; la diagonale passa il 12-40% della perdita naked.
- Capital scaling 600->2-5k: unico vincolo binding = cap $300/asset
  (a 5k book al 49% del target) -> AL DEPOSITO alzare a equity/2;
  min_order $5 lasciare; XS01 ~20k confermata; aspettativa onesta
  de-luckata 2k ~EUR 0.6-0.8/g, 5k ~EUR 1.4-2/g.

Nessun nuovo sleeve, book live invariato. 168 test verdi.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-02 23:31:38 +00:00

415 lines
19 KiB
Python
Raw Blame History

"""r0702_ell_rangecycle — Test ONESTO del claim "Elliott/Larry Williams range-cycle".
CLAIM (da video didattico, operazionalizzato SENZA conteggio soggettivo di onde):
(A) un segmento di trend che inizia con un impulso a RANGE COMPRESSO tende a produrre
un'estensione successiva AMPIA ("onda 3 larga");
(B) un segmento che inizia con impulso a RANGE AMPIO produce una fase finale TRONCATA
(ultimo terzo debole -> uscire prima).
Entrambi i claim, come operazionalizzati, predicono rho(impulso, seguito) NEGATIVA:
A: rho(range_impulso_norm, ritorno_normalizzato del RESTO del segmento) < 0
B: rho(range_impulso_norm, ritorno_normalizzato dell'ULTIMO TERZO) < 0
PRIOR ART (sweep 104-famiglie 2026-06-20): la compressione come *setup di breakout* è
già stata testata e NON regge — BRK08 (NR7), BRK10 (BB-squeeze long), CMB05 (squeeze->
breakout), VOL12 (low-vol timing). Qui la domanda è DIVERSA e a livello di SEGMENTO:
il range dell'impulso iniziale predice l'ampiezza/troncatura del seguito? Statistica
PRIMA della strategia: si strategizza SOLO se il segnale regge Bonferroni sul pre-2025.
METODO:
- Dati certificati BTC/ETH 1d (al.get) + settimanali costruiti dal 1d con resample
"7D", label='left', closed='left', origin ESPLICITA (7 ancore, banda riportata).
Niente DatetimeIndex.view("int64") (lezione pandas 2026-07-01).
- 2 definizioni MECCANICHE di segmento:
(i) TSMOM-flip: run massimali di sign(close[t]-close[t-L]) (L=30 bar 1d, 4 bar 1w).
Causale sia in start sia in end.
(ii) zigzag causale k*ATR: pivot = estremo confermato solo dopo retracement k*ATR
(k=3.0/ATR14 su 1d, k=2.0/ATR8 su 1w — FISSATI a priori, non cercati).
NB: il pivot è noto solo DOPO (lag di conferma) -> le statistiche sono
event-aligned ex-post (test di ASSOCIAZIONE); un overlay tradabile dovrebbe
scontare il lag. L'impulso (prime N barre) precede comunque il seguito ->
nessuna sovrapposizione meccanica impulso/seguito.
- Impulso = mean(TrueRange prime N barre) / ATR[pre-segmento], N in {3,5,10}.
- Seguito A = dir*(close_end - close_impulso) / (ATR_norm * sqrt(n_resto))
Seguito B = dir*(close_end - close_2/3) / (ATR_norm * sqrt(n_ultimo_terzo))
Normalizzazione PRIMARIA con ATR pre-segmento (stesso denominatore di impulso ->
bias comune-divisore POSITIVO su rho = CONSERVATIVO vs claim che predicono rho<0).
Sensibilità: anche ATR post-impulso (ANTI-conservativa: un impulso ampio gonfia
l'ATR post -> sgonfia il seguito -> fabbrica rho<0). Riportate entrambe, si testa
la primaria.
- Test: Spearman pooled BTC+ETH con NULL PERMUTATO (shuffle seguito<->impulso ENTRO
asset, 2000 permutazioni, two-sided). Secondario: differenza terzili (compresso -
ampio, terzili entro asset). Bonferroni su 24 celle primarie (2 def x 2 TF x 3 N x
2 claim; weekly = ancora lunedì, le altre 6 ancore = banda robustezza).
- Campione PICCOLO per costruzione (decine di segmenti, weekly ~390 barre) -> n
riportato per cella; p marginali NON vanno creduti.
Output: tabella celle + CSV nello scratchpad (ell_a_cells.csv). Strategizzazione
(study_marginal / study_family_honest) SOLO se una cella passa Bonferroni con il segno
giusto sul campione pre-2025.
"""
import sys
sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/alt")
import numpy as np
import pandas as pd
import altlib as al
SCRATCH = ("/tmp/claude-1001/-opt-docker-PythagorasGoal/"
"e00896d3-d4bb-4f2a-b471-55a1d88a12ba/scratchpad")
SEED = 20260702
N_PERM = 2000
HOLDOUT = pd.Timestamp("2025-01-01", tz="UTC")
N_LIST = (3, 5, 10)
# Parametri di segmentazione FISSATI A PRIORI (nessuna ricerca su di essi):
TSMOM_L = {"1d": 30, "1w": 4} # 30 giorni / 4 settimane (~28g)
ZZ_K = {"1d": 3.0, "1w": 2.0} # soglia retracement in multipli di ATR
ZZ_ATRWIN = {"1d": 14, "1w": 8}
BONF_M = 24 # 2 def x 2 TF x 3 N x 2 claim (weekly=ancora lun)
# ---------------------------------------------------------------------------
# Dati
# ---------------------------------------------------------------------------
def daily(asset: str) -> pd.DataFrame:
df = al.get(asset, "1d")
return df[["datetime", "open", "high", "low", "close", "volume"]].copy()
def weekly_from_daily(dfd: pd.DataFrame, anchor_day: int) -> pd.DataFrame:
"""Barre settimanali dal 1d. anchor_day=0 -> settimane che iniziano lunedì
(origin=2018-01-01, un lunedì); anchor_day=k sposta l'origine di k giorni.
resample('168h', label='left', closed='left', origin=esplicita) — '168h' e non
'7D': in pandas 2.x 'D' non è Tick-like e IGNOREREBBE origin (warning silenzioso
-> tutte le ancore identiche). Bins parziali (<7 giorni) scartati."""
origin = pd.Timestamp("2018-01-01", tz="UTC") + pd.Timedelta(days=anchor_day)
x = dfd.set_index(pd.DatetimeIndex(pd.to_datetime(dfd["datetime"], utc=True)))
g = x.resample("168h", label="left", closed="left", origin=origin)
w = g.agg(open=("open", "first"), high=("high", "max"), low=("low", "min"),
close=("close", "last"), volume=("volume", "sum"))
cnt = g["close"].count()
w = w[cnt >= 7].dropna().reset_index(names="datetime")
return w
def true_range(df: pd.DataFrame) -> np.ndarray:
h, l, c = (df[k].values.astype(float) for k in ("high", "low", "close"))
pc = np.roll(c, 1)
pc[0] = c[0]
return np.maximum(h - l, np.maximum(np.abs(h - pc), np.abs(l - pc)))
# ---------------------------------------------------------------------------
# Definizioni di segmento (meccaniche)
# ---------------------------------------------------------------------------
def tsmom_segments(close: np.ndarray, L: int) -> list[tuple[int, int, int]]:
"""Run massimali di sign(close[i]-close[i-L]). Ritorna (start, end, dir) inclusivi.
Causale: il segno a i usa solo close<=i; start noto allo start, end noto all'end."""
n = len(close)
s = np.zeros(n)
s[L:] = np.sign(close[L:] - close[:-L])
for i in range(1, n): # 0 -> segno precedente (nessun flip)
if s[i] == 0:
s[i] = s[i - 1]
segs, start = [], L
for i in range(L + 1, n + 1):
if i == n or s[i] != s[start]:
if s[start] != 0:
segs.append((start, i - 1, int(s[start])))
start = i
return segs
def zigzag_segments(df: pd.DataFrame, k: float, atr_win: int) -> list[tuple[int, int, int]]:
"""Zigzag causale a soglia k*ATR: pivot (estremo) confermato solo quando il prezzo
ritraccia k*ATR[i] dall'estremo corrente. Segmento = (pivot_prec+1 .. pivot), dir =
segno del leg. ⚠️ Il pivot è noto solo alla CONFERMA (lag): qui usato solo per
statistiche di associazione ex-post, non come segnale tradabile."""
c = df["close"].values.astype(float)
a = al.atr(df, atr_win)
n = len(c)
piv: list[int] = []
hi, hi_i, lo, lo_i, d = c[0], 0, c[0], 0, 0
for i in range(1, n):
if d >= 0:
if c[i] > hi:
hi, hi_i = c[i], i
if d == 0 and c[i] < lo:
lo, lo_i = c[i], i
if hi - c[i] > k * a[i]:
piv.append(hi_i)
d = -1
lo, lo_i = c[i], i
continue
if d == 0 and c[i] - lo > k * a[i]:
piv.append(lo_i)
d = 1
hi, hi_i = c[i], i
else:
if c[i] < lo:
lo, lo_i = c[i], i
if c[i] - lo > k * a[i]:
piv.append(lo_i)
d = 1
hi, hi_i = c[i], i
segs = []
for j in range(1, len(piv)):
p, q = piv[j - 1], piv[j]
if q > p + 1:
segs.append((p + 1, q, int(np.sign(c[q] - c[p]))))
return [s for s in segs if s[2] != 0]
# ---------------------------------------------------------------------------
# Misure per segmento
# ---------------------------------------------------------------------------
def measure(df: pd.DataFrame, segs, N: int, atr_win: int):
"""Per ogni segmento: impulso (range prime N barre / ATR pre-segmento) e seguito
per claim A (resto) e B (ultimo terzo). Normalizzazione primaria = ATR pre-segmento
(conservativa); sensibilità = ATR post-impulso. Ritorna DataFrame righe-segmento."""
c = df["close"].values.astype(float)
a = al.atr(df, atr_win)
tr = true_range(df)
ts = pd.to_datetime(df["datetime"], utc=True).values
rows = []
for (s, e, d) in segs:
seg_len = e - s + 1
if s - 1 < atr_win or seg_len < N + 3:
continue
a_pre = a[s - 1]
if not np.isfinite(a_pre) or a_pre <= 0:
continue
imp = float(np.mean(tr[s:s + N]) / a_pre)
i_end = s + N - 1 # ultima barra dell'impulso
a_post = a[i_end] if a[i_end] > 0 else np.nan
# Claim A: resto del segmento (da fine impulso a fine segmento), in dir. trend
n_rest = e - i_end
move_a = d * (c[e] - c[i_end])
seqA_pre = float(move_a / (a_pre * np.sqrt(n_rest)))
seqA_post = float(move_a / (a_post * np.sqrt(n_rest))) if np.isfinite(a_post) else np.nan
# ampiezza massima favorevole del resto (descrittiva)
if d > 0:
mfe = np.max(df["high"].values[i_end + 1:e + 1]) - c[i_end]
else:
mfe = c[i_end] - np.min(df["low"].values[i_end + 1:e + 1])
mfe_pre = float(mfe / (a_pre * np.sqrt(n_rest)))
# Claim B: ultimo terzo del segmento (deve iniziare dopo l'impulso)
l0 = s + int(np.ceil(seg_len * 2.0 / 3.0)) # prima barra dell'ultimo terzo
seqB_pre = seqB_post = np.nan
if l0 - 1 >= i_end and e - (l0 - 1) >= 2:
n_last = e - (l0 - 1)
move_b = d * (c[e] - c[l0 - 1])
a_b = a[l0 - 1]
seqB_pre = float(move_b / (a_pre * np.sqrt(n_last)))
seqB_post = float(move_b / (a_b * np.sqrt(n_last))) if a_b > 0 else np.nan
rows.append(dict(t0=ts[s], seg_len=seg_len, dir=d, imp=imp,
seqA=seqA_pre, seqA_post=seqA_post, mfe=mfe_pre,
seqB=seqB_pre, seqB_post=seqB_post))
return pd.DataFrame(rows)
# ---------------------------------------------------------------------------
# Statistica: Spearman pooled + null permutato entro-asset; terzili
# ---------------------------------------------------------------------------
def _spearman(x: np.ndarray, y: np.ndarray) -> float:
rx = pd.Series(x).rank().values
ry = pd.Series(y).rank().values
sx, sy = rx.std(), ry.std()
if sx == 0 or sy == 0:
return 0.0
return float(np.corrcoef(rx, ry)[0, 1])
def _tercile_diff(imp, seq, aid):
"""mean(seguito | terzile compresso) - mean(seguito | terzile ampio), terzili
per-asset (>=9 segmenti per asset, altrimenti asset escluso). Claim -> diff>0."""
lo_v, hi_v = [], []
for a in np.unique(aid):
m = aid == a
if m.sum() < 9:
continue
q1, q2 = np.quantile(imp[m], [1 / 3, 2 / 3])
lo_v.append(seq[m & (imp <= q1)])
hi_v.append(seq[m & (imp >= q2)])
if not lo_v:
return np.nan
return float(np.concatenate(lo_v).mean() - np.concatenate(hi_v).mean())
def perm_test(imp, seq, aid, n_perm=N_PERM, seed=SEED):
"""Shuffle del seguito rispetto all'impulso ENTRO asset. p two-sided su Spearman
(primario) e su tercile diff (secondario)."""
ok = np.isfinite(imp) & np.isfinite(seq)
imp, seq, aid = imp[ok], seq[ok], aid[ok]
n = len(imp)
if n < 12:
return dict(rho=np.nan, p=np.nan, ter=np.nan, p_ter=np.nan, n=n)
rho = _spearman(imp, seq)
ter = _tercile_diff(imp, seq, aid)
rng = np.random.default_rng(seed)
masks = [aid == a for a in np.unique(aid)]
cnt_r = cnt_t = 0
n_t = 0
for _ in range(n_perm):
sp = seq.copy()
for m in masks:
sp[m] = rng.permutation(sp[m])
if abs(_spearman(imp, sp)) >= abs(rho):
cnt_r += 1
if np.isfinite(ter):
tp = _tercile_diff(imp, sp, aid)
if np.isfinite(tp):
n_t += 1
if abs(tp) >= abs(ter):
cnt_t += 1
p = (1 + cnt_r) / (n_perm + 1)
p_ter = (1 + cnt_t) / (n_t + 1) if (np.isfinite(ter) and n_t > 0) else np.nan
return dict(rho=round(rho, 3), p=p, ter=(round(ter, 3) if np.isfinite(ter) else np.nan),
p_ter=p_ter, n=n)
# ---------------------------------------------------------------------------
# Celle
# ---------------------------------------------------------------------------
def build_frames(tf: str, anchor: int):
out = {}
for a in ("BTC", "ETH"):
d = daily(a)
out[a] = d if tf == "1d" else weekly_from_daily(d, anchor)
return out
def segments_for(df, sdef: str, tf: str):
if sdef == "tsmom":
return tsmom_segments(df["close"].values.astype(float), TSMOM_L[tf])
return zigzag_segments(df, ZZ_K[tf], ZZ_ATRWIN[tf])
def cell_rows(frames, sdef: str, tf: str, N: int) -> pd.DataFrame:
parts = []
for i, (a, df) in enumerate(frames.items()):
m = measure(df, segments_for(df, sdef, tf), N, ZZ_ATRWIN[tf])
if len(m):
m["aid"] = i
m["asset"] = a
parts.append(m)
return pd.concat(parts, ignore_index=True) if parts else pd.DataFrame()
def run_cell(rows: pd.DataFrame, claim: str, pre2025: bool, norm: str = "pre"):
if rows.empty:
return dict(rho=np.nan, p=np.nan, ter=np.nan, p_ter=np.nan, n=0)
r = rows
if pre2025:
r = r[pd.to_datetime(r["t0"], utc=True) < HOLDOUT]
col = {"A": {"pre": "seqA", "post": "seqA_post"},
"B": {"pre": "seqB", "post": "seqB_post"}}[claim][norm]
return perm_test(r["imp"].values.astype(float), r[col].values.astype(float),
r["aid"].values)
def main():
print(__doc__.split("\n")[0])
print(f"seed={SEED} n_perm={N_PERM} Bonferroni m={BONF_M} (celle primarie)")
results = []
frames_1d = build_frames("1d", 0)
frames_1w = {k: build_frames("1w", k) for k in range(7)}
for a, df in frames_1d.items():
print(f" {a} 1d: {len(df)} barre {df['datetime'].iloc[0]} -> {df['datetime'].iloc[-1]}"
f" | 1w(lun): {len(frames_1w[0][a])} barre")
for sdef in ("tsmom", "zigzag"):
for tf in ("1d", "1w"):
fr = frames_1d if tf == "1d" else frames_1w[0]
# conteggio segmenti grezzi (informativo)
nseg = {a: len(segments_for(df, sdef, tf)) for a, df in fr.items()}
print(f"\n[{sdef} {tf}] segmenti grezzi: {nseg}")
for N in N_LIST:
rows = cell_rows(fr, sdef, tf, N)
for claim in ("A", "B"):
full = run_cell(rows, claim, pre2025=False)
ins = run_cell(rows, claim, pre2025=True)
post = run_cell(rows, claim, pre2025=False, norm="post")
rec = dict(sdef=sdef, tf=tf, N=N, claim=claim,
n=full["n"], rho=full["rho"], p=full["p"],
p_bonf=min(1.0, full["p"] * BONF_M) if np.isfinite(full["p"]) else np.nan,
ter=full["ter"], p_ter=full["p_ter"],
rho_is=ins["rho"], p_is=ins["p"], n_is=ins["n"],
rho_postnorm=post["rho"])
# banda ancore weekly (tutte e 7, stessa cella)
if tf == "1w":
rhos, ps = [], []
for k in range(7):
rr = run_cell(cell_rows(frames_1w[k], sdef, tf, N), claim, False)
rhos.append(float(rr["rho"]) if np.isfinite(rr["rho"]) else np.nan)
ps.append(float(rr["p"]) if np.isfinite(rr["p"]) else np.nan)
if np.any(np.isfinite(rhos)):
med = float(np.nanmedian(rhos))
rec["anchor_rho_band"] = (round(float(np.nanmin(rhos)), 3),
round(med, 3),
round(float(np.nanmax(rhos)), 3))
rec["anchor_p_band"] = (round(float(np.nanmin(ps)), 4),
round(float(np.nanmax(ps)), 4))
rec["anchor_sign_flips"] = int(np.nansum(
np.sign(rhos) != np.sign(med)))
else:
rec["anchor_rho_band"] = rec["anchor_p_band"] = None
rec["anchor_sign_flips"] = None
results.append(rec)
R = pd.DataFrame(results)
R.to_csv(f"{SCRATCH}/ell_a_cells.csv", index=False)
print("\n" + "=" * 118)
print("TABELLA CELLE (norm primaria = ATR pre-segmento; claim A/B predicono rho<0; p permutato two-sided)")
print("=" * 118)
hdr = (f"{'def':7}{'tf':4}{'N':3}{'cl':3}{'n':5}{'rho':>7}{'p':>8}{'p_bonf':>8}"
f"{'ter(c-a)':>9}{'p_ter':>8}{'rho_IS':>8}{'p_IS':>8}{'n_IS':>5}{'rho_post':>9} anchors(1w)")
print(hdr)
for _, r in R.iterrows():
anch = ""
if r["tf"] == "1w" and r["anchor_rho_band"] is not None:
anch = (f"rho[min/med/max]={r['anchor_rho_band']} "
f"p[{r['anchor_p_band'][0]}..{r['anchor_p_band'][1]}] "
f"sign_flips={r['anchor_sign_flips']}")
print(f"{r['sdef']:7}{r['tf']:4}{r['N']:<3}{r['claim']:3}{r['n']:<5}"
f"{r['rho']:>7}{r['p']:>8.4f}{r['p_bonf']:>8.3f}"
f"{str(r['ter']):>9}{r['p_ter']:>8.4f}{str(r['rho_is']):>8}{r['p_is']:>8.4f}{r['n_is']:>5}"
f"{str(r['rho_postnorm']):>9} {anch}")
# ------------------------------------------------------------------
# Verdetti
# ------------------------------------------------------------------
print("\n" + "=" * 118)
for claim in ("A", "B"):
sub = R[R["claim"] == claim]
sig = sub[(sub["p_bonf"] < 0.05) & (sub["rho"] < 0)]
sig_is = sub[(sub["p_is"] * BONF_M < 0.05) & (sub["rho_is"] < 0)]
neg = int((sub["rho"] < 0).sum())
print(f"CLAIM {claim}: celle={len(sub)} rho<0 (segno del claim)={neg}/{len(sub)} "
f"pass Bonferroni FULL={len(sig)} pass Bonferroni pre-2025={len(sig_is)}")
if len(sig) == 0 and len(sig_is) == 0:
print(f" -> NESSUNA cella regge Bonferroni. Claim {claim} NON supportato: "
f"si strategizza solo con evidenza statistica -> STOP qui per questo claim.")
else:
print(f" -> celle sopravvissute:\n{sig.to_string()}")
print(" -> procedere a strategizzazione SOLO su selezione pre-2025 "
"(study_family_honest).")
n_med = int(R["n"].median())
print(f"\nCAVEAT CAMPIONE: n mediano per cella = {n_med} segmenti "
f"(weekly ~{int(R[R['tf'] == '1w']['n'].median())}). Con decine di osservazioni "
"un p marginale (0.01-0.05 raw) è indistinguibile dal multiple-testing: "
"24 celle primarie + 2 claim correlati + 7 ancore weekly.")
print("NOTE: (1) zigzag = associazione ex-post (pivot noto solo a conferma); "
"(2) norm 'post' (ATR post-impulso) è ANTI-conservativa (divisore contaminato "
"dall'impulso) -> se rho<0 appare solo lì, è artefatto di normalizzazione; "
"(3) daily = ancora 00:00 UTC fissa (feed certificato).")
print(f"CSV: {SCRATCH}/ell_a_cells.csv")
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink