Files
Adriano Dal Pastro cff5fa2bf5 research(sweep): 5 thread paralleli — 0 nuovi sleeve, STATARB-RESID LEAD ortogonale+eseguibile
Ricerca onesta su aree inesplorate (harness altlib+xsec_v2_nonmom, tutti i gate incl.
study_family_honest anti-selection-on-holdout). Branch main, nessun impatto live, test 143/143.

1 XSEC low-risk cousins (MAX/idio-vol/Amihud) -> 1 LEAD (IVOL), STAT-MODE, DSR 0.37<0.95
2 XSEC momentum-structure vs XS01            -> tutto REDUNDANT (sostituire XS01 distrugge hold)
3 Meta-allocazione dinamica (4 sleeve)       -> pesi fissi vincono (gia quasi risk-parity)
4 Segnali ortogonali ETH/BTC (2 gambe)       -> STATARB-RESID + DVOLSPREAD LEAD
5 1-gamba a segnale (MACD/RSI/Supertrend/...) -> 0/12 earns_slot (trend=TP01, MR morta, hedge)

LEAD principale STATARB-RESID (mean-rev residuo ETH-b*BTC, OLS rolling, 2 gambe): primo stream
INSIEME ortogonale (corr->book 0.027, beta-mkt 0.013) ED eseguibile a $600 (haircut ~0, NON
STAT-MODE) -> cadono i 2 muri di XS01/opzioni. Resta solo il muro dell'edge (Sharpe 0.84,
DSR 0.929 same-sign <0.95). Causalita+fee verificate dal coordinatore. Forward-monitor, non sleeve.

Soffitto direzionale ~1.3 riconfermato. Diario 2026-06-29-strategy-search-5threads.md, CLAUDE.md agg.

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

401 lines
17 KiB
Python

"""signal_inout_1leg — 1-GAMBA a SEGNALI classici (entrata+uscita) su BTC/ETH (2026-06-29).
TESI
----
Un filone con un VANTAGGIO STRUTTURALE diverso dai book multi-gamba (XS01, CC01, basis,
short-vol) che restano STAT-MODE perche' a $600 non si eseguono: qui ogni strategia e' a
**1 SOLA GAMBA** (un singolo asset, BTC o ETH) con entrata E uscita gestite da SEGNALI
CLASSICI (MACD, RSI, Supertrend/ATR-trail, Donchian, Bollinger, EMA-cross, MACD+ADX).
Turnover basso => realmente ESEGUIBILE a $600 (cap $300/asset, min $5). Il vantaggio del
filone NON e' lo Sharpe assoluto — e' l'eseguibilita'.
MA la lezione del progetto e' brutale: un 1-gamba direzionale su BTC/ETH o e'
(a) TREND-FOLLOWING -> corr ~0.7-0.9 a TP01, marginal REDUNDANT (TP01 travestito), oppure
(b) MEAN-REVERSION -> morto sul feed reale (fade negativo anche a fee zero, v2.0.0), o
muore di fee a sub-daily.
Il soffitto direzionale BTC/ETH e' ~1.3 (= TP01). Quindi NON si giudica sullo Sharpe
assoluto ma sul MARGINALE vs TP01 (earns_slot), e la cella si sceglie IN-SAMPLE-ONLY
(no peeking del hold-out — il punto cieco del filone B / intraday-ERM).
GATE (tutti dall'harness condiviso altlib, leak-free by construction)
---------------------------------------------------------------------
1. study_family_honest(name, factory, grid, tfs):
- sceglie la cella per Sharpe IN-SAMPLE (pre-2025), MAI per max hold-out;
- study_marginal sulla cella scelta (corr vs TP01, blend uplift full/hold,
is_hedge, has_insample_edge, robust_oos multi-cut) -> earns_slot;
- deflated-Sharpe (Bailey & Lopez de Prado) su TUTTE le celle del grid (multiple-testing);
- earns_slot_honest = earns_slot AND DSR>=0.95.
2. causality_ok: la costruzione del segnale e' causale (max_tail_diff ~0, no peeking).
3. eval_weights_smallcap(capital=600, min_order=5): HAIRCUT $600 + n. trade eseguiti reali
(il punto FORTE del filone — turnover basso = eseguibile davvero).
4. fee-sweep 0.00-0.20% RT (dentro study_weights) a frequenza reale.
5. day_boundary_robust: per completezza (segnali di prezzo => INVARIANT atteso).
Decisione finale per ogni segnale:
SLEEVE-CANDIDATE-eseguibile -> earns_slot_honest=True AND haircut ~0 (executable). RARO.
LEAD-forward-monitor -> marginal ADDS ma DSR/hold corto, o edge esile: monitor.
REDUNDANT-vs-TP01 -> trend travestito (corr alta, uplift ~0).
SCARTATO -> abs FAIL (negativo), DILUTES, o morte per fee.
USO: cd /opt/docker/PythagorasGoal && uv run python scripts/research/signal_inout_1leg.py
Idempotente, solo stdout. NON committa (lo fa il coordinatore).
"""
from __future__ import annotations
import sys
import numpy as np
import pandas as pd
sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/alt")
import altlib as al # noqa: E402
CERTIFIED = al.CERTIFIED
TFS = ("1d", "12h", "8h")
# ===========================================================================
# INDICATORI extra (causali) non gia' in altlib
# ===========================================================================
def macd_lines(c: np.ndarray, fast: int, slow: int, sig: int):
"""MACD = EMA(fast) - EMA(slow); signal = EMA(MACD, sig). EMA adjust=False => causale."""
macd = al.ema(c, fast) - al.ema(c, slow)
signal = al.ema(macd, sig)
return macd, signal
def supertrend_dir(df: pd.DataFrame, atr_win: int, mult: float) -> np.ndarray:
"""Supertrend classico -> direzione {+1 up, -1 down}. CAUSALE: dir[i] usa close[i] e le
bande finali calcolate fino a i-1 (mai high/low di i come prezzo d'ingresso)."""
h = df["high"].values.astype(float)
l = df["low"].values.astype(float)
c = df["close"].values.astype(float)
a = al.atr(df, atr_win)
hl2 = (h + l) / 2.0
upper = hl2 + mult * a
lower = hl2 - mult * a
n = len(c)
fu = np.zeros(n)
fl = np.zeros(n)
d = np.ones(n)
for i in range(n):
if i == 0 or not np.isfinite(a[i]):
fu[i], fl[i], d[i] = upper[i], lower[i], 1.0
continue
fu[i] = upper[i] if (upper[i] < fu[i - 1] or c[i - 1] > fu[i - 1]) else fu[i - 1]
fl[i] = lower[i] if (lower[i] > fl[i - 1] or c[i - 1] < fl[i - 1]) else fl[i - 1]
if c[i] > fu[i - 1]:
d[i] = 1.0
elif c[i] < fl[i - 1]:
d[i] = -1.0
else:
d[i] = d[i - 1]
return d
def adx(df: pd.DataFrame, win: int = 14) -> np.ndarray:
"""ADX di Wilder (forza di trend), causale via EWM. Usato come gate anti-whipsaw."""
h = df["high"].values.astype(float)
l = df["low"].values.astype(float)
up = np.zeros(len(h))
dn = np.zeros(len(h))
up[1:] = h[1:] - h[:-1]
dn[1:] = l[:-1] - l[1:]
plus_dm = np.where((up > dn) & (up > 0), up, 0.0)
minus_dm = np.where((dn > up) & (dn > 0), dn, 0.0)
atr_ = al.atr(df, win)
den = np.where(atr_ > 0, atr_, np.nan)
plus_di = 100 * pd.Series(plus_dm).ewm(alpha=1 / win, adjust=False).mean().values / den
minus_di = 100 * pd.Series(minus_dm).ewm(alpha=1 / win, adjust=False).mean().values / den
s = plus_di + minus_di
dx = 100 * np.abs(plus_di - minus_di) / np.where(s > 0, s, np.nan)
return pd.Series(dx).ewm(alpha=1 / win, adjust=False).mean().values
def _hold(pos_raw: np.ndarray) -> np.ndarray:
"""ffill dello stato fra i segnali (NaN = mantieni posizione precedente), start flat."""
return pd.Series(pos_raw).ffill().fillna(0.0).values
# ===========================================================================
# FACTORY dei segnali — ognuna: factory(tf, **params) -> target_fn(df)->posizione causale
# (tf passa solo per firma; la granularita' la sceglie candidate_daily caricando get(asset,tf))
# ===========================================================================
def make_macd(mode: str):
def factory(tf, fast, slow, sig):
def target(df):
c = df["close"].values.astype(float)
m, s = macd_lines(c, fast, slow, sig)
if mode == "LF":
return np.where(m > s, 1.0, 0.0)
return np.where(m > s, 1.0, -1.0)
return target
return factory
def make_rsi():
"""Mean-reversion long-flat: entra quando RSI<oversold, esce quando RSI>overbought, HOLD in mezzo."""
def factory(tf, win, oversold, overbought):
def target(df):
c = df["close"].values.astype(float)
ind = al.rsi(c, win)
raw = np.full(len(c), np.nan)
raw[ind < oversold] = 1.0
raw[ind > overbought] = 0.0
return _hold(raw)
return target
return factory
def make_supertrend(mode: str):
def factory(tf, atr_win, mult):
def target(df):
d = supertrend_dir(df, atr_win, mult)
return np.clip(d, 0, None) if mode == "LF" else d
return target
return factory
def make_donchian(mode: str):
"""Turtle: long su breakout del canale alto, esce/short sul canale basso, HOLD in mezzo."""
def factory(tf, win):
def target(df):
hi, lo = al.donchian(df, win)
c = df["close"].values.astype(float)
raw = np.full(len(c), np.nan)
raw[c > hi] = 1.0
raw[c < lo] = 0.0 if mode == "LF" else -1.0
return _hold(raw)
return target
return factory
def make_bbands(mode: str):
"""MR: long sotto banda bassa, esce al ritorno sopra la media. BO: long sopra banda alta,
flat/short sotto banda bassa."""
def factory(tf, win, k):
def target(df):
c = df["close"].values.astype(float)
up, mid, lo = al.bbands(c, win, k)
raw = np.full(len(c), np.nan)
if mode == "MR":
raw[c < lo] = 1.0
raw[c > mid] = 0.0
else: # breakout
raw[c > up] = 1.0
raw[c < lo] = 0.0
return _hold(raw)
return target
return factory
def make_ema(mode: str):
def factory(tf, fast, slow):
def target(df):
c = df["close"].values.astype(float)
ef, es = al.ema(c, fast), al.ema(c, slow)
if mode == "LF":
return np.where(ef > es, 1.0, 0.0)
return np.where(ef > es, 1.0, -1.0)
return target
return factory
def make_macd_adx():
"""MACD long-flat gated da ADX (entra long solo se MACD>signal E trend abbastanza forte)."""
def factory(tf, fast, slow, sig, adx_win, adx_thr):
def target(df):
c = df["close"].values.astype(float)
m, s = macd_lines(c, fast, slow, sig)
a = adx(df, adx_win)
return np.where((m > s) & (a > adx_thr), 1.0, 0.0)
return target
return factory
# ===========================================================================
# GRID
# ===========================================================================
def macd_grid():
g = []
for fast in (8, 12, 16):
for slow in (21, 26, 34):
for sig in (9,):
if fast < slow:
g.append(dict(fast=fast, slow=slow, sig=sig))
return g
RSI_GRID = [dict(win=14, oversold=os, overbought=ob)
for os in (25, 30, 35) for ob in (60, 65, 70)]
SUPER_GRID = [dict(atr_win=w, mult=m) for w in (10, 14, 20) for m in (2.0, 2.5, 3.0)]
DONCH_GRID = [dict(win=w) for w in (20, 30, 40, 55)]
BB_GRID = [dict(win=20, k=k) for k in (2.0, 2.5)]
EMA_GRID = [dict(fast=f, slow=s) for f in (10, 20, 30) for s in (50, 100, 200) if f < s]
MACD_ADX_GRID = [dict(fast=12, slow=26, sig=9, adx_win=14, adx_thr=t) for t in (15, 20, 25)]
# ===========================================================================
# DRIVER
# ===========================================================================
def _yr_line(absolute: dict) -> str:
"""Per-anno minimo-fra-asset dalla parte assoluta (cella scelta), TF best."""
cells = absolute.get("cells", [])
if not cells:
return ""
c = cells[0]
out = []
for a, pa in c["per_asset"].items():
yr = " ".join(f"{y}:{(d['ret'] if isinstance(d, dict) else d) * 100:+.0f}%"
for y, d in pa["yearly"].items())
out.append(f" {a}: {yr}")
return "\n".join(out)
def classify(rep: dict, haircut_ok: bool) -> tuple[str, str]:
"""Verdetto a 4 etichette + motivo di una riga."""
if rep.get("chosen") is None:
return "SCARTATO", "nessuna cella in-sample valida"
m = rep["marginal"]["marginal"]
mv = rep["marginal"]["marginal_verdict"]
abs_grade = rep["marginal"]["abs_grade"]
corr = m.get("corr_full")
uph = (m.get("blends", {}).get("w25", {}) or {}).get("uplift_hold")
hsh = m.get("cand_hold_sharpe")
trend_like = corr is not None and corr >= 0.5
if rep.get("earns_slot_honest"):
if haircut_ok:
return "SLEEVE-CANDIDATE-eseguibile", "earns_slot_honest=True + haircut $600 ~0 (extra-scettico: possibile selection/fee artifact)"
return "LEAD-forward-monitor", "earns_slot_honest=True ma haircut $600 non trascurabile"
if mv == "HEDGE":
return "LEAD-forward-monitor", "HEDGE: low-corr ma paga solo quando TP01 e' debole (non alpha standing); DSR/abs sotto soglia"
if mv == "DILUTES":
return "SCARTATO", "DILUTES: trascina giu' il blend TP01 (no edge marginale)"
if abs_grade == "FAIL":
if trend_like:
return "REDUNDANT-vs-TP01", f"trend = TP01 travestito (corr {corr}); la cella in-sample-best (sub-daily) overfitta -> hold-Sh {hsh} OOS"
return "SCARTATO", f"abs FAIL: la cella in-sample-best non generalizza OOS (hold-Sh {hsh})"
if mv == "REDUNDANT" or trend_like:
return "REDUNDANT-vs-TP01", f"trend travestito: corr {corr} a TP01, marginal {mv}, uplift-hold non persistente"
if mv == "ADDS":
return "LEAD-forward-monitor", "marginal ADDS ma deflated-Sharpe non passa (multiple-testing)"
return "REDUNDANT-vs-TP01", f"{mv}: nessun uplift marginale robusto (corr {corr})"
def run_family(name: str, factory, grid, tfs=TFS):
print("\n" + "=" * 100)
print(f"### {name} (grid {len(grid)} celle x {len(tfs)} TF = {len(grid) * len(tfs)} prove)")
print("=" * 100)
rep = al.study_family_honest(name, factory, grid, tfs)
ch = rep.get("chosen")
if ch is None:
print(f" -> {rep.get('reason', 'no chosen')}")
v, why = classify(rep, False)
print(f" VERDETTO: {v}{why}")
return rep, (name, v, why, None)
fn = factory(tf=ch["tf"], **ch["params"])
m = rep["marginal"]["marginal"]
mv = rep["marginal"]["marginal_verdict"]
absr = rep["marginal"]["absolute"]
bl = m.get("blends", {})
w25 = bl.get("w25", {})
w50 = bl.get("w50", {})
print(f" cella scelta IN-SAMPLE: tf={ch['tf']} {ch['params']} "
f"insample-Sh={ch['insample_sharpe']} full-Sh={ch['full_sharpe']}")
print(f" ABS grade={rep['marginal']['abs_grade']} | cand full-Sh={m.get('cand_full_sharpe')} "
f"hold-Sh={m.get('cand_hold_sharpe')} (TP01 full {m.get('tp01_full_sharpe')}/hold {m.get('tp01_hold_sharpe')})")
print(f" MARGINAL={mv} corr->TP01 full={m.get('corr_full')} hold={m.get('corr_hold')} "
f"is_hedge={m.get('is_hedge')} has_insample_edge={m.get('has_insample_edge')} robust_oos={m.get('robust_oos')}")
print(f" blend w25: full {w25.get('full')} (uplift {w25.get('uplift_full')}) "
f"hold {w25.get('hold')} (uplift {w25.get('uplift_hold')}) DD {w25.get('dd')}")
print(f" blend w50: full {w50.get('full')} (uplift {w50.get('uplift_full')}) "
f"hold {w50.get('hold')} (uplift {w50.get('uplift_hold')})")
print(f" multicut uplift {m.get('multicut_uplift')} persistent={m.get('multicut_persistent')}")
print(f" DEFLATED-SHARPE={rep.get('deflated_sharpe')} (null-max {rep.get('expected_null_max')}, "
f"n_cells {rep.get('n_cells')}) dsr_pass={rep.get('dsr_pass')}")
print(f" >>> earns_slot(marginal)={rep['earns_slot_marginal']} EARNS_SLOT_HONEST={rep['earns_slot_honest']}")
# causalita'
caus = al.causality_ok(fn, tf=ch["tf"])
print(f" causality_ok={caus['ok']} max_tail_diff={caus['max_tail_diff']}")
# HAIRCUT $600 + n. trade (il punto forte del filone)
print(" ESEGUIBILITA' $600 (haircut + n.trade eseguiti reali):")
haircut_ok = True
for a in CERTIFIED:
df = al.get(a, ch["tf"])
tgt = al._call_target(fn, df, a)
sc = al.eval_weights_smallcap(df, tgt, capital=600, min_order=5)
print(f" {a}: modeled Sh={sc['modeled']['sharpe']} real$600 Sh={sc['realistic']['sharpe']} "
f"haircut={sc['sharpe_haircut']} n_trade_eseguiti={sc['n_executed_trades']} "
f"turnover/yr={sc['executed_turnover_per_year']}")
if abs(sc["sharpe_haircut"]) > 0.25:
haircut_ok = False
# day-boundary (segnale di prezzo => INVARIANT atteso)
try:
db = al.day_boundary_robust(fn, tf=ch["tf"])
print(f" day_boundary: {db['verdict']} (spread {db.get('spread')})")
except Exception as e: # noqa
print(f" day_boundary: skip ({type(e).__name__})")
# per-anno
yr = _yr_line(absr)
if yr:
print(" per-anno (cella scelta):")
print(yr)
v, why = classify(rep, haircut_ok)
print(f" VERDETTO: {v}{why}")
return rep, (name, v, why, rep.get("earns_slot_honest"))
def main():
print("#" * 100)
print("# SIGNAL-IN/OUT 1-GAMBA — MACD/RSI/Supertrend/Donchian/Bollinger/EMA/MACD+ADX su BTC,ETH")
print(f"# TP01 baseline daily full Sharpe = {round(al._sh(al.tp01_baseline_daily()), 3)} (il soffitto da battere al MARGINE)")
print(f"# TF testati: {TFS} | fee 0.10% RT + sweep 0..0.20% | capitale reale $600")
print("#" * 100)
summary = []
families = [
("MACD-LF (long-flat)", make_macd("LF"), macd_grid()),
("MACD-LS (long-short)", make_macd("LS"), macd_grid()),
("RSI-MR (mean-rev long-flat)", make_rsi(), RSI_GRID),
("SUPERTREND-LF", make_supertrend("LF"), SUPER_GRID),
("SUPERTREND-LS", make_supertrend("LS"), SUPER_GRID),
("DONCHIAN-LF (turtle)", make_donchian("LF"), DONCH_GRID),
("DONCHIAN-LS (turtle)", make_donchian("LS"), DONCH_GRID),
("BBANDS-MR (mean-rev)", make_bbands("MR"), BB_GRID),
("BBANDS-BO (breakout)", make_bbands("BO"), BB_GRID),
("EMA-CROSS-LF", make_ema("LF"), EMA_GRID),
("EMA-CROSS-LS", make_ema("LS"), EMA_GRID),
("MACD+ADX-LF (gated)", make_macd_adx(), MACD_ADX_GRID),
]
for nm, fac, grid in families:
try:
_, row = run_family(nm, fac, grid)
summary.append(row)
except Exception as e: # noqa
import traceback
print(f"\n[ERRORE in {nm}] {type(e).__name__}: {e}")
traceback.print_exc()
summary.append((nm, "ERRORE", str(e), None))
print("\n" + "#" * 100)
print("# SOMMARIO FINALE")
print("#" * 100)
for nm, v, why, esh in summary:
flag = " <<< ESEGUIBILE+SCORRELATO" if esh else ""
print(f" {nm:32s} -> {v}{flag}")
print(f" {why}")
any_slot = any(esh for *_, esh in summary)
print("\nCONCLUSIONE: c'e' un 1-gamba a segnale che AGGIUNGE oltre TP01 ED e' eseguibile a $600?")
print(f" -> {'SI (verificare extra-scetticismo: selection/fee artifact)' if any_slot else 'NO — tutto REDUNDANT (trend=TP01) o SCARTATO (MR morta / fee). Risultato valido: base-rate confermata.'}")
if __name__ == "__main__":
main()