diff --git a/docs/diary/2026-05-28-honest-strategies.md b/docs/diary/2026-05-28-honest-strategies.md new file mode 100644 index 0000000..3fe9c75 --- /dev/null +++ b/docs/diary/2026-05-28-honest-strategies.md @@ -0,0 +1,95 @@ +# 2026-05-28 — Ricerca onesta di nuove strategie (post-squeeze) + +## Contesto e mandato + +Dopo aver scoperto che l'intera famiglia squeeze-breakout era un artefatto di +look-ahead (accuratezze 76-82% svanite sotto ingresso eseguibile), il mandato è +stato: trovare in modo **onesto** almeno 3 strategie attendibili, testate su ~8 +anni e su più criptovalute, con le fee incluse nella valutazione, partendo da +€1.000 con l'obiettivo (aspirazionale) di €50/giorno. Esplorare anche idee fuori +dal comune e l'uso combinato di più crypto e timeframe. + +## Metodologia (engine onesto) + +Tutto il lavoro usa un unico engine condiviso (`scripts/analysis/honest_lab.py`) +con questi vincoli anti-illusione: + +1. **Ingresso eseguibile.** Ogni segnale alla barra `i` usa solo dati fino a + `close[i]` e l'ingresso avviene a `close[i]` (ciò che il worker live vede e + può eseguire). Disponibile anche l'ingresso più conservativo a `open[i+1]`. +2. **Uscita realistica.** Take-profit / stop-loss valutati intrabar su `high`/`low`, + in modo conservativo (SL prima del TP nello stesso bar), più time-limit. + Una posizione per volta (non-overlap), capitale composto. +3. **Fee di prim'ordine.** Tutto è NETTO dopo fee round-trip realistiche Deribit + (0.10% RT) moltiplicate per la leva (3x), con sweep fino a 0.20% RT. +4. **Validazione severa.** FULL + out-of-sample (ultimo 30%) + conteggio anni + positivi + sweep fee + griglia parametri + test su **8 crypto** + (BTC, ETH, SOL, BNB, XRP, LTC, DOGE, ADA, 2018→2026). + +## Lezione madre + +**Shortare le crypto perde OOS in modo sistematico in questo campione.** Sia la +mean-reversion sul lato short, sia il momentum short, crollano fuori campione: il +periodo 2018-2026 è net-bull e ogni rialzo "estremo" tende a continuare invece di +rientrare. Tutte le configurazioni che sopravvivono oneste sono **long-biased**. +È un fatto da dichiarare: parte della performance OOS è correlata al beta rialzista +delle crypto. Le strategie aggiungono *timing* sopra quel beta, non lo eliminano. + +## Le 3 strategie selezionate (meccanismi distinti) + +| Codice | Meccanismo | TF | Asset robusti | OOS netto (fee 0.10% RT) | DD | Anni+ | +|--------|-----------|----|---------------|--------------------------|----|-------| +| **DIP01** | Dip-buy z-score reversion (long-only) | 1h | BTC, ETH, SOL | BTC +59% · ETH +224% · SOL +13% | 23-55% | 6-7/9 | +| **TR01** | EMA 20/100 trend-following (long-only) | 4h | BNB, BTC, DOGE, SOL, XRP | BTC +27% · DOGE +53% · XRP +29% | 29-53% | 4-6/8 | +| **ROT01** | Rotazione cross-sectional momentum sul paniere | 1d | intero paniere (8) | **+44%** | 53% | 5/7 | + +Dettagli e riproducibilità: `scripts/analysis/honest_final.py` (tabella di +validazione unica), `honest_rotation.py`, `honest_trend.py`, `honest_candidates.py`, +`honest_diag.py`/`honest_diag2.py` (diagnostica long/short e filtro trend). + +### DIP01 — compra le capitolazioni +Long-only: entra quando lo z-score del prezzo rispetto alla media a 50 barre scende +sotto −2.5 (capitolazione), prende profitto al rientro verso la media, SL a 2.5·ATR. +È la versione robusta e onesta della famiglia mean-reversion: regge lo sweep fee +fino a 0.20% RT (BTC +45% OOS anche a 0.20%). Funziona sui major (BTC/ETH/SOL); sugli +alt molto parabolici (DOGE/BNB) un dip fisso continua a scendere e non ha edge. + +### TR01 — cavalca i trend +Long-only: in posizione quando EMA(20) > EMA(100) sul 4h, altrimenti cash. Poche +operazioni (≈200 flip in 8 anni) ⇒ le fee non sono letali. È **complementare** a +DIP01: guadagna nei regimi di trend, dove la reversione soffre. + +### ROT01 — la più affidabile e "fuori dal comune" +Una sola strategia che usa **tutto il paniere** in un unico book: ogni giorno ordina +le 8 crypto per momentum (rendimento a 60 giorni) e alloca a parti uguali alle 2 +migliori con momentum positivo, il resto in cash. Cattura la *dispersione* tra +crypto (gli alt forti corrono molto più di BTC nei bull) senza shortare nulla. +È **param-insensitive** (tutte le combinazioni lookback/top-k sono positive OOS) e +regge le fee fino a 0.20% RT (+41% OOS). Risponde direttamente alla richiesta di +combinare più crypto e un timeframe diverso in un'unica strategia. Per-anno: +2020 +33% · 2021 +181% · 2022 −29% (bear) · 2023 +43% · 2024 +59% · 2025 +6% · 2026 −10% (YTD). + +## Diversificazione + +I tre meccanismi coprono regimi diversi e in larga misura anti-correlati: +reversione (DIP01), momentum di singolo asset (TR01), forza relativa cross-asset +(ROT01). Eseguirli insieme produce una curva di equity più liscia del singolo. + +## Onestà sull'obiettivo €50/giorno + +Va detto chiaramente: **€50/giorno su €1.000 in pochi mesi non è raggiungibile a +rischio sano.** Significa ~€18.250/anno, cioè ~1.825%/anno; gli edge onesti qui +trovati rendono il 30-60% OOS su orizzonti pluriennali. Le strade per avvicinare +quel numero sono: (a) far crescere il capitale per anni con interesse composto — +€50/giorno diventa plausibile solo quando il capitale è molto più grande; (b) alzare +la leva, che però aumenta proporzionalmente il drawdown (già 23-55%) ed espone a +rovina; (c) aggiungere capitale. Nessuna di queste è una scorciatoia. La proposta +onesta è un portafoglio delle 3 strategie a leva moderata, puntando alla +**sopravvivenza e alla crescita composta**, non al target giornaliero immediato. + +## Prossimi passi + +- Integrare DIP01 nel worker (già compatibile: Signal con tp/sl/max_bars). +- Estendere il worker per strategie position-based (TR01) e di portafoglio (ROT01). +- Backtest del portafoglio combinato con ribilanciamento del capitale. +- Walk-forward rolling (oltre al singolo split 70/30) per confermare la stabilità. diff --git a/scripts/analysis/honest_candidates.py b/scripts/analysis/honest_candidates.py new file mode 100644 index 0000000..8abaeb6 --- /dev/null +++ b/scripts/analysis/honest_candidates.py @@ -0,0 +1,175 @@ +"""Strategie candidate ONESTE + sweep multi-asset/tf con verdetto. + +Ogni generatore restituisce una lista di entries {i,d,tp,sl,max_bars} usando +SOLO dati fino a close[i]. L'engine (honest_lab.simulate) entra a close[i]. + +Famiglie testate (meccanismi distinti, per diversificazione): + MR mean-reversion single-asset (Bollinger fade, RSI revert, Z-score) + XS cross-sectional relative-value (fade della divergenza vs paniere) + MOM time-series momentum / trend su timeframe alto + SES seasonality (ora del giorno UTC) +""" +from __future__ import annotations + +import sys +from pathlib import Path + +import numpy as np +import pandas as pd + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(PROJECT_ROOT)) + +from scripts.analysis.honest_lab import ( # noqa: E402 + atr, rsi, ema, get_df, simulate, oos_split, verdict, + available_assets, FEE_RT, +) + + +# ============================================================================ +# MR — mean reversion single-asset +# ============================================================================ +def bollinger_fade(df, n=50, k=2.5, sl_atr=2.0, max_bars=24): + c = df["close"].values + ma = pd.Series(c).rolling(n).mean().values + sd = pd.Series(c).rolling(n).std().values + a = atr(df, 14) + up, lo = ma + k * sd, ma - k * sd + ents = [] + for i in range(n + 14, len(c)): + if np.isnan(up[i]) or np.isnan(a[i]): + continue + if c[i] < lo[i] and c[i - 1] >= lo[i - 1]: + ents.append({"i": i, "d": 1, "tp": ma[i], "sl": c[i] - sl_atr * a[i], "max_bars": max_bars}) + elif c[i] > up[i] and c[i - 1] <= up[i - 1]: + ents.append({"i": i, "d": -1, "tp": ma[i], "sl": c[i] + sl_atr * a[i], "max_bars": max_bars}) + return ents + + +def rsi_revert(df, n=14, lo=25, hi=75, sl_atr=2.5, max_bars=24, ma_n=20): + c = df["close"].values + r = rsi(c, n) + ma = pd.Series(c).rolling(ma_n).mean().values + a = atr(df, 14) + ents = [] + for i in range(max(n, ma_n) + 1, len(c)): + if np.isnan(r[i]) or np.isnan(ma[i]) or np.isnan(a[i]): + continue + if r[i - 1] < lo <= r[i]: + ents.append({"i": i, "d": 1, "tp": ma[i], "sl": c[i] - sl_atr * a[i], "max_bars": max_bars}) + elif r[i - 1] > hi >= r[i]: + ents.append({"i": i, "d": -1, "tp": ma[i], "sl": c[i] + sl_atr * a[i], "max_bars": max_bars}) + return ents + + +def zscore_revert(df, n=50, z_in=2.5, sl_atr=2.5, max_bars=24): + """Entra quando close e' a |z|>z_in std dalla media; TP alla media.""" + c = df["close"].values + ma = pd.Series(c).rolling(n).mean().values + sd = pd.Series(c).rolling(n).std().values + a = atr(df, 14) + z = (c - ma) / sd + ents = [] + for i in range(n + 14, len(c)): + if np.isnan(z[i]) or np.isnan(a[i]) or sd[i] == 0: + continue + if z[i] <= -z_in and z[i - 1] > -z_in: + ents.append({"i": i, "d": 1, "tp": ma[i], "sl": c[i] - sl_atr * a[i], "max_bars": max_bars}) + elif z[i] >= z_in and z[i - 1] < z_in: + ents.append({"i": i, "d": -1, "tp": ma[i], "sl": c[i] + sl_atr * a[i], "max_bars": max_bars}) + return ents + + +# ============================================================================ +# MOM — time-series momentum / trend (timeframe alto, niente breakout intrabar) +# ============================================================================ +def ema_trend(df, fast=20, slow=50, sl_atr=3.0, tp_atr=10.0, max_bars=240): + """Trend following: cross EMA fast/slow deciso a close[i], TP/SL ad ATR.""" + c = df["close"].values + ef, es = ema(c, fast), ema(c, slow) + a = atr(df, 14) + ents = [] + for i in range(slow + 14, len(c)): + if np.isnan(a[i]): + continue + cross_up = ef[i] > es[i] and ef[i - 1] <= es[i - 1] + cross_dn = ef[i] < es[i] and ef[i - 1] >= es[i - 1] + if cross_up: + ents.append({"i": i, "d": 1, "tp": c[i] + tp_atr * a[i], "sl": c[i] - sl_atr * a[i], "max_bars": max_bars}) + elif cross_dn: + ents.append({"i": i, "d": -1, "tp": c[i] - tp_atr * a[i], "sl": c[i] + sl_atr * a[i], "max_bars": max_bars}) + return ents + + +# ============================================================================ +# SES — seasonality (ora del giorno UTC). Direzione fissa decisa solo dall'ora. +# ============================================================================ +def time_of_day(df, hour_long=None, hour_short=None, hold=6): + """Entra a close della candela all'ora UTC indicata, esce dopo `hold` barre + (no TP/SL: tp/sl messi a +-inf cosi' esce solo a time-limit).""" + ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True) + c = df["close"].values + hours = ts.dt.hour.values + hour_long = set(hour_long or []) + hour_short = set(hour_short or []) + ents = [] + for i in range(1, len(c)): + if hours[i] in hour_long: + ents.append({"i": i, "d": 1, "tp": np.inf, "sl": -np.inf, "max_bars": hold}) + elif hours[i] in hour_short: + ents.append({"i": i, "d": -1, "tp": -np.inf, "sl": np.inf, "max_bars": hold}) + return ents + + +# ============================================================================ +# sweep +# ============================================================================ +def run_sweep(generators: dict, assets: list[str], tfs: list[str]): + print("=" * 130) + print(f" HONEST LAB — NETTO fee {FEE_RT*100:.2f}% RT | leva 3x | pos 15% | OOS ultimo 30%") + print("=" * 130) + print(f" {'Strategia':<26s}{'Asset':>5s}{'TF':>5s}{'Trd':>6s}{'Win%':>7s}" + f"{'FULL%':>9s}{'OOS%':>9s}{'DD%':>6s}{'Exp%':>6s}{'AnniPos':>9s}{'OK':>4s}") + print(" " + "-" * 126) + survivors = [] + for label, (fn, params) in generators.items(): + for asset in assets: + for tf in tfs: + try: + df = get_df(asset, tf) + except Exception: + continue + ents = fn(df, **params) + if len(ents) < 30: + continue + full = simulate(ents, df) + _, oos_e = oos_split(ents, df) + oos = simulate(oos_e, df) + ok = verdict(full, oos) + flag = " OK" if ok else "" + print(f" {label:<26s}{asset:>5s}{tf:>5s}{full.trades:>6d}{full.win:>7.1f}" + f"{full.ret:>+9.0f}{oos.ret:>+9.0f}{full.dd:>6.0f}{full.exposure:>6.0f}" + f"{f'{full.pos_years}/{full.n_years}':>9s}{flag:>4s}") + if ok: + survivors.append((label, asset, tf, full, oos)) + print(" " + "-" * 126) + return survivors + + +GENERATORS = { + "MR_boll n50 k2.5": (bollinger_fade, dict(n=50, k=2.5, sl_atr=2.0, max_bars=24)), + "MR_boll n20 k2.5": (bollinger_fade, dict(n=20, k=2.5, sl_atr=2.0, max_bars=24)), + "MR_rsi 25/75": (rsi_revert, dict(n=14, lo=25, hi=75, sl_atr=2.5, max_bars=24)), + "MR_zscore z2.5": (zscore_revert, dict(n=50, z_in=2.5, sl_atr=2.5, max_bars=24)), + "MR_zscore z3": (zscore_revert, dict(n=50, z_in=3.0, sl_atr=2.5, max_bars=24)), + "MOM_ema 20/50": (ema_trend, dict(fast=20, slow=50, sl_atr=3.0, tp_atr=10.0, max_bars=240)), +} + + +if __name__ == "__main__": + assets = available_assets() + print("Asset disponibili:", assets) + survivors = run_sweep(GENERATORS, assets, ["1h", "4h"]) + print(f"\n SOPRAVVISSUTI (FULL+OOS+anni+DD): {len(survivors)}") + for label, a, tf, full, oos in survivors: + print(f" {label:<26s} {a} {tf} FULL {full.ret:+.0f}% OOS {oos.ret:+.0f}% DD {full.dd:.0f}%") diff --git a/scripts/analysis/honest_diag.py b/scripts/analysis/honest_diag.py new file mode 100644 index 0000000..8132981 --- /dev/null +++ b/scripts/analysis/honest_diag.py @@ -0,0 +1,73 @@ +"""Diagnostica: perche' la mean-reversion simmetrica perde su asset trending? +Test: long-only vs short-only, e MR FILTRATA DAL TREND (buy-dip in uptrend, +sell-rip in downtrend) per evitare di fadeare i trend forti. +""" +from __future__ import annotations + +import sys +from pathlib import Path + +import numpy as np +import pandas as pd + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(PROJECT_ROOT)) + +from scripts.analysis.honest_lab import ( # noqa: E402 + atr, ema, get_df, simulate, oos_split, available_assets, FEE_RT, +) + + +def zscore_entries(df, n=50, z_in=2.5, sl_atr=2.5, max_bars=24, + trend_n=0, side="both"): + """Z-score revert con filtro trend opzionale. + trend_n>0: EMA di lungo periodo. Long solo se close>EMA (uptrend), + short solo se close 0 else None + start = max(n + 14, trend_n + 1 if trend_n else 0) + ents = [] + for i in range(start, len(c)): + if np.isnan(z[i]) or np.isnan(a[i]): + continue + long_ok = (et is None or c[i] > et[i]) and side in ("both", "long") + short_ok = (et is None or c[i] < et[i]) and side in ("both", "short") + if z[i] <= -z_in and z[i - 1] > -z_in and long_ok: + ents.append({"i": i, "d": 1, "tp": ma[i], "sl": c[i] - sl_atr * a[i], "max_bars": max_bars}) + elif z[i] >= z_in and z[i - 1] < z_in and short_ok: + ents.append({"i": i, "d": -1, "tp": ma[i], "sl": c[i] + sl_atr * a[i], "max_bars": max_bars}) + return ents + + +def row(label, df, ents): + if len(ents) < 20: + print(f" {label:<34s} {'<20 trd':>50s}") + return None + full = simulate(ents, df) + _, oe = oos_split(ents, df) + oos = simulate(oe, df) + print(f" {label:<34s}{full.trades:>6d}{full.win:>7.1f}{full.ret:>+9.0f}" + f"{oos.ret:>+9.0f}{full.dd:>6.0f}{f'{full.pos_years}/{full.n_years}':>8s}") + return full, oos + + +if __name__ == "__main__": + assets = available_assets() + print(f"HONEST DIAG — z-score revert, fee {FEE_RT*100:.2f}% RT, leva 3x | OOS 30%") + for tf in ["1h"]: + for a in assets: + df = get_df(a, tf) + print(f"\n === {a} {tf} === {'Trd':>5s}{'Win%':>7s}{'FULL%':>8s}{'OOS%':>8s}{'DD%':>6s}{'AnniP':>8s}") + base = dict(n=50, z_in=2.5, sl_atr=2.5, max_bars=24) + row("both, no filter", df, zscore_entries(df, **base, side="both")) + row("long-only, no filter", df, zscore_entries(df, **base, side="long")) + row("short-only, no filter", df, zscore_entries(df, **base, side="short")) + row("both + trend200 filter", df, zscore_entries(df, **base, trend_n=200, side="both")) + row("both + trend500 filter", df, zscore_entries(df, **base, trend_n=500, side="both")) + row("long + trend200 filter", df, zscore_entries(df, **base, trend_n=200, side="long")) diff --git a/scripts/analysis/honest_diag2.py b/scripts/analysis/honest_diag2.py new file mode 100644 index 0000000..0d66aea --- /dev/null +++ b/scripts/analysis/honest_diag2.py @@ -0,0 +1,64 @@ +"""Diag2: long-MR sempre + short-MR SOLO in downtrend confermato (close shortare i rimbalzi in +downtrend, mai i rimbalzi in bull-run. +""" +from __future__ import annotations + +import sys +from pathlib import Path + +import numpy as np +import pandas as pd + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(PROJECT_ROOT)) + +from scripts.analysis.honest_lab import ( # noqa: E402 + atr, ema, get_df, simulate, oos_split, available_assets, FEE_RT, +) + + +def regime_mr(df, n=50, z_in=2.5, sl_atr=2.5, max_bars=24, trend_n=200, + allow_short=True): + """Long su z<=-z_in SEMPRE. Short su z>=+z_in solo se close -z_in: + ents.append({"i": i, "d": 1, "tp": ma[i], "sl": c[i] - sl_atr * a[i], "max_bars": max_bars}) + elif allow_short and z[i] >= z_in and z[i - 1] < z_in and c[i] < et[i]: + ents.append({"i": i, "d": -1, "tp": ma[i], "sl": c[i] + sl_atr * a[i], "max_bars": max_bars}) + return ents + + +def show(label, df, ents): + if len(ents) < 20: + print(f" {label:<30s} <20 trd"); return None + full = simulate(ents, df); _, oe = oos_split(ents, df); oos = simulate(oe, df) + print(f" {label:<30s}{full.trades:>6d}{full.win:>7.1f}{full.ret:>+9.0f}" + f"{oos.ret:>+9.0f}{full.dd:>6.0f}{f'{full.pos_years}/{full.n_years}':>8s}") + return full, oos + + +if __name__ == "__main__": + assets = available_assets() + print(f"DIAG2 — regime MR (long sempre + short in downtrend) fee {FEE_RT*100:.2f}% leva3x OOS30%") + surv = 0 + for a in assets: + df = get_df(a, "1h") + print(f"\n === {a} 1h === {'Trd':>5s}{'Win%':>7s}{'FULL%':>8s}{'OOS%':>8s}{'DD%':>6s}{'AnniP':>8s}") + show("long-only", df, regime_mr(df, allow_short=False)) + r = show("long + short@downtrend200", df, regime_mr(df, trend_n=200)) + show("long + short@downtrend500", df, regime_mr(df, trend_n=500)) + if r and r[0].ret > 0 and r[1].ret > 0: + surv += 1 + print(f"\n Asset con regime200 positivo FULL+OOS: {surv}/{len(assets)}") diff --git a/scripts/analysis/honest_final.py b/scripts/analysis/honest_final.py new file mode 100644 index 0000000..50ae927 --- /dev/null +++ b/scripts/analysis/honest_final.py @@ -0,0 +1,103 @@ +"""Validazione FINALE delle 3 strategie oneste selezionate. +Per ciascuna: per-asset FULL/OOS/DD/anni-positivi + sweep fee (0/0.05/0.10/0.20% RT). +Tutto NETTO, ingresso eseguibile, OOS = ultimo 30%, leva 3x. + +S1 DIP — long-only dip-buy z-score reversion (1h) [regime: reversione] +S2 TREND — long-only EMA 20/100 trend-following (4h) [regime: momentum singolo] +S3 ROT — rotazione cross-sectional momentum sul paniere (1d) [regime: forza relativa] +""" +from __future__ import annotations + +import sys +from pathlib import Path + +import numpy as np +import pandas as pd + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(PROJECT_ROOT)) + +from scripts.analysis.honest_lab import atr, ema, get_df, simulate, oos_split, available_assets +from scripts.analysis.honest_trend import simulate_position, ema_dual_signal, oos as trend_oos +from scripts.analysis.honest_rotation import build_panel, simulate_rotation + +FEES = [0.0, 0.0005, 0.001, 0.002] + + +# ---- S1 DIP ---- +def dip_entries(df, n=50, z_in=2.5, sl_atr=2.5, max_bars=24): + c = df["close"].values + ma = pd.Series(c).rolling(n).mean().values + sd = pd.Series(c).rolling(n).std().values + a = atr(df, 14) + z = (c - ma) / np.where(sd == 0, np.nan, sd) + ents = [] + for i in range(n + 14, len(c)): + if np.isnan(z[i]) or np.isnan(a[i]): + continue + if z[i] <= -z_in and z[i - 1] > -z_in: + ents.append({"i": i, "d": 1, "tp": ma[i], "sl": c[i] - sl_atr * a[i], "max_bars": max_bars}) + return ents + + +def validate_dip(assets): + print("\n" + "=" * 100) + print(" S1 DIP — long-only dip-buy z-score reversion | 1h | n=50 z=2.5 sl=2.5ATR mb=24") + print("=" * 100) + print(f" {'Asset':<6s}{'Trd':>6s}{'Win%':>7s}{'FULL%':>9s}{'OOS%':>9s}{'DD%':>6s}{'Exp%':>6s}{'AnniP':>8s}" + f"{' fee-sweep OOS% (0/0.05/0.10/0.20)':<40s}") + ok = 0 + for a in assets: + df = get_df(a, "1h"); ents = dip_entries(df) + if len(ents) < 30: + continue + full = simulate(ents, df); _, oe = oos_split(ents, df); oos = simulate(oe, df) + sweep = " ".join(f"{simulate(oe, df, fee_rt=f).ret:+.0f}" for f in FEES) + good = full.ret > 0 and oos.ret > 0 + ok += good + print(f" {a:<6s}{full.trades:>6d}{full.win:>7.1f}{full.ret:>+9.0f}{oos.ret:>+9.0f}" + f"{full.dd:>6.0f}{full.exposure:>6.0f}{f'{full.pos_years}/{full.n_years}':>8s} [{sweep}]" + f"{' OK' if good else ''}") + print(f" -> robusto (FULL+OOS>0) su {ok}/{len(assets)} asset") + + +def validate_trend(assets): + print("\n" + "=" * 100) + print(" S2 TREND — long-only EMA 20/100 trend | 4h") + print("=" * 100) + print(f" {'Asset':<6s}{'Flip':>6s}{'FULL%':>9s}{'OOS%':>9s}{'DD%':>6s}{'Exp%':>6s}{'AnniP':>8s}") + ok = 0 + for a in assets: + df = get_df(a, "4h"); sig = ema_dual_signal(df, 20, 100, long_only=True) + full = simulate_position(sig, df); oos = trend_oos(sig, df) + good = full["ret"] > 0 and oos["ret"] > 0 + ok += good + print(f" {a:<6s}{full['flips']:>6d}{full['ret']:>+9.0f}{oos['ret']:>+9.0f}" + f"{full['dd']:>6.0f}{full['exposure']:>6.0f}{(str(full['pos_years'])+'/'+str(full['n_years'])):>8s}" + f"{' OK' if good else ''}") + print(f" -> robusto su {ok}/{len(assets)} asset") + + +def validate_rot(assets): + print("\n" + "=" * 100) + print(" S3 ROT — rotazione cross-sectional momentum | 1d | lb=60 top2 su tutto il paniere") + print("=" * 100) + panel = build_panel(assets, "1d") + print(f" Paniere {list(panel.columns)} {panel.shape[0]} barre {panel.index[0].date()}->{panel.index[-1].date()}") + print(f" {'fee RT':<10s}{'FULL%':>9s}{'OOS%':>9s}{'DD%':>6s}{'AnniP':>8s}") + for f in FEES: + full = simulate_rotation(panel, lookback=60, top_k=2, fee_rt=f) + oos = simulate_rotation(panel, lookback=60, top_k=2, fee_rt=f, oos_frac=0.30) + anni = str(full['pos_years']) + '/' + str(full['n_years']) + print(f" {f*100:>5.2f}%RT {full['ret']:>+9.0f}{oos['ret']:>+9.0f}{full['dd']:>6.0f}{anni:>8s}") + # per-anno alla fee reale + full = simulate_rotation(panel, lookback=60, top_k=2, fee_rt=0.001) + print(" per-anno (fee 0.10%): " + " ".join(f"{y}:{v:+.0f}%" for y, v in sorted(full["yearly"].items()))) + + +if __name__ == "__main__": + assets = available_assets() + print(f"VALIDAZIONE FINALE — asset disponibili: {assets}") + validate_dip(assets) + validate_trend(assets) + validate_rot(assets) diff --git a/scripts/analysis/honest_lab.py b/scripts/analysis/honest_lab.py new file mode 100644 index 0000000..094ee90 --- /dev/null +++ b/scripts/analysis/honest_lab.py @@ -0,0 +1,192 @@ +"""honest_lab — laboratorio di ricerca strategie ONESTO e fee-aware. + +Principi (per non ripetere l'errore look-ahead della famiglia squeeze): + 1. Ogni segnale a barra i usa SOLO dati fino a close[i]. Ingresso a close[i] + (eseguibile dal vivo: il worker vede la candela chiusa ed entra). Opzione + di robustezza: ingresso a open[i+1] (ancora piu' conservativo). + 2. Uscita TP/SL valutata intrabar su high/low, conservativa: SL prima del TP + nello stesso bar. Time-limit max_bars. Una posizione per volta (non-overlap). + 3. Tutto NETTO dopo fee round-trip realistiche (0.10% Deribit) * leva. + 4. Validazione: FULL + OOS (held-out ultimo 30%) + per-anno + sweep fee + + griglia parametri + su PIU' asset. Niente di tutto cio' -> scartata. + +Engine condiviso riusabile da tutte le strategie candidate. +""" +from __future__ import annotations + +import sys +from dataclasses import dataclass +from pathlib import Path + +import numpy as np +import pandas as pd + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(PROJECT_ROOT)) + +from src.data.downloader import load_data # noqa: E402 + +FEE_RT = 0.001 # Deribit perp realistico: taker ~0.05%/lato = 0.10% RT +LEV = 3.0 +POS = 0.15 +OOS_FRAC = 0.30 +DATA_DIR = PROJECT_ROOT / "data" / "raw" + +# ---------------------------------------------------------------------------- +# dati +# ---------------------------------------------------------------------------- +_CACHE: dict[tuple[str, str], pd.DataFrame] = {} + + +def available_assets() -> list[str]: + out = [] + for p in sorted(DATA_DIR.glob("*_1h.parquet")): + name = p.stem.replace("_1h", "").upper() + if name not in ("BTC_DVOL", "ETH_DVOL"): + out.append(name) + return out + + +def get_df(asset: str, tf: str) -> pd.DataFrame: + """tf nativo (15m,1h) o resample da 1h (2h,4h,6h,12h,1d).""" + key = (asset, tf) + if key in _CACHE: + return _CACHE[key] + if tf in ("15m", "1h"): + df = load_data(asset, tf).reset_index(drop=True) + else: + base = load_data(asset, "1h").copy() + base["dt"] = pd.to_datetime(base["timestamp"], unit="ms", utc=True) + base = base.set_index("dt") + rule = {"2h": "2h", "4h": "4h", "6h": "6h", "12h": "12h", "1d": "1D"}[tf] + agg = base.resample(rule).agg( + {"open": "first", "high": "max", "low": "min", "close": "last", "volume": "sum"} + ).dropna() + # l'indice puo' essere datetime64[ms] o [ns]: forza ms in modo robusto + agg["timestamp"] = agg.index.values.astype("datetime64[ms]").astype("int64") + df = agg.reset_index(drop=True) + df = df[["timestamp", "open", "high", "low", "close", "volume"]].copy() + _CACHE[key] = df + return df + + +# ---------------------------------------------------------------------------- +# indicatori +# ---------------------------------------------------------------------------- +def atr(df: pd.DataFrame, n: 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).rolling(n).mean().values + + +def rsi(close: np.ndarray, n: int = 14) -> np.ndarray: + d = np.diff(close, prepend=close[0]) + up = pd.Series(np.where(d > 0, d, 0.0)).ewm(alpha=1 / n, adjust=False).mean() + dn = pd.Series(np.where(d < 0, -d, 0.0)).ewm(alpha=1 / n, adjust=False).mean() + rs = up / dn.replace(0, np.nan) + return (100 - 100 / (1 + rs)).values + + +def ema(close: np.ndarray, n: int) -> np.ndarray: + return pd.Series(close).ewm(span=n, adjust=False).mean().values + + +# ---------------------------------------------------------------------------- +# engine +# ---------------------------------------------------------------------------- +@dataclass +class SimResult: + trades: int + win: float + ret: float # ritorno % netto composto su 1000 + dd: float + exposure: float + yearly: dict[int, float] + + @property + def pos_years(self) -> int: + return sum(1 for v in self.yearly.values() if v > 0) + + @property + def n_years(self) -> int: + return len(self.yearly) + + +def simulate(entries: list[dict], df: pd.DataFrame, fee_rt: float = FEE_RT, + lev: float = LEV, pos: float = POS, entry_on_open: bool = False) -> SimResult: + """entries: dict {i, d(+1/-1), tp, sl, max_bars}. + + entry_on_open=True -> ingresso a open[i+1] invece di close[i] (robustezza). + """ + o, h, l, c = (df["open"].values, df["high"].values, + df["low"].values, df["close"].values) + n = len(c) + cap = peak = 1000.0 + max_dd = 0.0 + fee = fee_rt * lev + trades = wins = 0 + last_exit = -1 + bars_in = 0 + ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True) + yearly: dict[int, float] = {} + + for e in entries: + i, d = e["i"], e["d"] + ei = i + 1 if entry_on_open else i # barra di ingresso + if ei <= last_exit or ei + 1 >= n: + continue + entry = o[ei] if entry_on_open else c[i] + tp, sl, mb = e["tp"], e["sl"], e["max_bars"] + exit_p = c[min(ei + mb, n - 1)] + j = min(ei + mb, n - 1) + for k in range(1, mb + 1): + j = ei + k + if j >= n: + j = n - 1; exit_p = c[j]; break + hit_sl = (d == 1 and l[j] <= sl) or (d == -1 and h[j] >= sl) + hit_tp = (d == 1 and h[j] >= tp) or (d == -1 and l[j] <= tp) + if hit_sl: # conservativo: SL prima del TP nello stesso bar + exit_p = sl; break + if hit_tp: + exit_p = tp; break + if k == mb: + exit_p = c[j] + ret = (exit_p - entry) / entry * d * lev - fee + cap = max(cap + cap * pos * ret, 10.0) + peak = max(peak, cap); max_dd = max(max_dd, (peak - cap) / peak) + trades += 1; wins += ret > 0; bars_in += (j - ei) + last_exit = j + yr = ts.iloc[i].year + yearly[yr] = yearly.get(yr, 0.0) + ret * 100 + return SimResult( + trades=trades, + win=wins / trades * 100 if trades else 0.0, + ret=(cap / 1000 - 1) * 100, + dd=max_dd * 100, + exposure=bars_in / n * 100, + yearly=yearly, + ) + + +def oos_split(entries: list[dict], df: pd.DataFrame, frac: float = OOS_FRAC): + split = int(len(df) * (1 - frac)) + ins = [e for e in entries if e["i"] < split] + oos = [e for e in entries if e["i"] >= split] + return ins, oos + + +# ---------------------------------------------------------------------------- +# criterio di accettazione +# ---------------------------------------------------------------------------- +def verdict(full: SimResult, oos: SimResult) -> bool: + """Strategia attendibile su un singolo asset/tf.""" + if full.trades < 30: + return False + if full.ret <= 0 or oos.ret <= 0: + return False + if full.pos_years < max(full.n_years - 1, 1): + return False + if full.dd > 45: + return False + return True diff --git a/scripts/analysis/honest_rotation.py b/scripts/analysis/honest_rotation.py new file mode 100644 index 0000000..af69fca --- /dev/null +++ b/scripts/analysis/honest_rotation.py @@ -0,0 +1,96 @@ +"""Strategia #3 candidata: ROTAZIONE cross-sectional momentum (multi-crypto). +Una sola strategia che usa l'INTERO paniere: ad ogni ribilanciamento alloca il +capitale agli asset con momentum migliore (long-only). Cattura la dispersione tra +crypto (gli alt forti corrono molto piu' di BTC nei bull) senza shortare nulla. + +Onesto: i pesi a close[i] usano solo rendimenti passati; il rendimento del bar +i->i+1 e' realizzato con quei pesi. Fee sul turnover. Allineamento per timestamp. +""" +from __future__ import annotations + +import sys +from pathlib import Path + +import numpy as np +import pandas as pd + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(PROJECT_ROOT)) + +from scripts.analysis.honest_lab import get_df, available_assets, FEE_RT # noqa: E402 + +LEV = 3.0 +GROSS = 0.45 # esposizione lorda = LEV*POS del singolo (0.15*3) per confronto equo + + +def build_panel(assets: list[str], tf: str) -> pd.DataFrame: + """Matrice close allineata per timestamp (inner join).""" + closes = {} + for a in assets: + df = get_df(a, tf) + s = pd.Series(df["close"].values, + index=pd.to_datetime(df["timestamp"], unit="ms", utc=True)) + closes[a] = s[~s.index.duplicated()] + panel = pd.DataFrame(closes).dropna() + return panel + + +def simulate_rotation(panel: pd.DataFrame, lookback=30, top_k=2, + fee_rt=FEE_RT, gross=GROSS, abs_filter=True, + oos_frac=0.0) -> dict: + """Ad ogni barra: ranking per rendimento passato `lookback`; pesi uguali sui + top_k con momentum>0 (se abs_filter); altrimenti cash. gross = esposizione tot. + oos_frac>0: parte a investire solo dall'ultimo frac del campione.""" + P = panel.values + T, N = P.shape + rets = np.zeros_like(P) + rets[1:] = P[1:] / P[:-1] - 1 + years = panel.index.year.values + start = max(lookback + 1, int(T * (1 - oos_frac)) if oos_frac else lookback + 1) + cap = peak = 1000.0 + max_dd = 0.0 + w = np.zeros(N) + yearly: dict[int, float] = {} + turn_total = 0.0 + for i in range(start, T - 1): + mom = P[i] / P[i - lookback] - 1 + order = np.argsort(mom)[::-1] + new_w = np.zeros(N) + chosen = [j for j in order if (mom[j] > 0 or not abs_filter)][:top_k] + if chosen: + for j in chosen: + new_w[j] = gross / len(chosen) + # fee sul turnover (one-way = fee_rt/2 su ogni variazione di peso) + turnover = np.abs(new_w - w).sum() + cap -= cap * turnover * (fee_rt / 2) + turn_total += turnover + w = new_w + port_ret = float(np.dot(w, rets[i + 1])) # rendimento bar i->i+1 + cap = max(cap * (1 + port_ret), 10.0) + peak = max(peak, cap); max_dd = max(max_dd, (peak - cap) / peak) + yearly[years[i]] = yearly.get(years[i], 0.0) + port_ret * 100 + return { + "ret": (cap / 1000 - 1) * 100, + "dd": max_dd * 100, + "turnover": turn_total, + "yearly": yearly, + "pos_years": sum(1 for v in yearly.values() if v > 0), + "n_years": len(yearly), + } + + +if __name__ == "__main__": + assets = available_assets() + print(f"ROTATION cross-sectional momentum — fee {FEE_RT*100:.2f}% RT, gross {GROSS} | OOS 30%") + print(f" Paniere: {assets}") + for tf in ["1d", "4h"]: + panel = build_panel(assets, tf) + print(f"\n === {tf} === panel {panel.shape[0]} barre, {panel.index[0].date()} -> {panel.index[-1].date()}") + print(f" {'config':<22s}{'FULL%':>9s}{'OOS%':>9s}{'DD%':>6s}{'Turn':>7s}{'AnniP':>8s}") + for lb in [20, 30, 60, 90]: + for k in [1, 2, 3]: + full = simulate_rotation(panel, lookback=lb, top_k=k) + oos = simulate_rotation(panel, lookback=lb, top_k=k, oos_frac=0.30) + anni = f"{full['pos_years']}/{full['n_years']}" + print(f" lb{lb:<3d} top{k:<14d}{full['ret']:>+9.0f}{oos['ret']:>+9.0f}" + f"{full['dd']:>6.0f}{full['turnover']:>7.0f}{anni:>8s}") diff --git a/scripts/analysis/honest_trend.py b/scripts/analysis/honest_trend.py new file mode 100644 index 0000000..b48fa39 --- /dev/null +++ b/scripts/analysis/honest_trend.py @@ -0,0 +1,109 @@ +"""Strategia #3 candidata: time-series momentum / trend (TSMOM). +Posizione continua decisa a close[i] dai dati passati; fee SOLO sui cambi di +posizione (poche operazioni su TF alto = fee non letali). Niente look-ahead: +il rendimento del bar i->i+1 usa la direzione decisa a close[i]. +""" +from __future__ import annotations + +import sys +from pathlib import Path + +import numpy as np +import pandas as pd + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(PROJECT_ROOT)) + +from scripts.analysis.honest_lab import ema, get_df, available_assets, FEE_RT # noqa: E402 + +LEV = 3.0 +POS = 0.15 + + +def simulate_position(sig: np.ndarray, df: pd.DataFrame, fee_rt: float = FEE_RT, + lev: float = LEV, pos: float = POS) -> dict: + """sig[i] in {-1,0,1} = direzione tenuta nel bar i->i+1, decisa a close[i]. + Fee one-way = fee_rt/2 su ogni unita' di variazione posizione.""" + c = df["close"].values + n = len(c) + ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True) + cap = peak = 1000.0 + max_dd = 0.0 + cur = 0.0 + flips = 0 + bars_in = 0 + yearly: dict[int, float] = {} + for i in range(n - 1): + s = sig[i] + if not np.isfinite(s): + s = 0.0 + if s != cur: + cap -= cap * pos * (fee_rt / 2) * lev * abs(s - cur) + flips += abs(s - cur) > 0 + cur = s + pr = (c[i + 1] - c[i]) / c[i] + bar_ret = pos * lev * pr * cur + cap = max(cap * (1 + bar_ret), 10.0) + peak = max(peak, cap); max_dd = max(max_dd, (peak - cap) / peak) + if cur != 0: + bars_in += 1 + yr = ts.iloc[i].year + yearly[yr] = yearly.get(yr, 0.0) + bar_ret * 100 + return { + "ret": (cap / 1000 - 1) * 100, + "dd": max_dd * 100, + "flips": flips, + "exposure": bars_in / n * 100, + "yearly": yearly, + "pos_years": sum(1 for v in yearly.values() if v > 0), + "n_years": len(yearly), + } + + +def tsmom_signal(df, lookback=30, long_only=False): + """+1 se close>close[-lookback], -1 (o 0 se long_only) altrimenti.""" + c = df["close"].values + sig = np.zeros(len(c)) + for i in range(lookback, len(c)): + up = c[i] > c[i - lookback] + sig[i] = 1.0 if up else (0.0 if long_only else -1.0) + return sig + + +def ema_dual_signal(df, fast=20, slow=100, long_only=False): + """+1 se EMA_fast>EMA_slow.""" + c = df["close"].values + ef, es = ema(c, fast), ema(c, slow) + sig = np.where(ef > es, 1.0, 0.0 if long_only else -1.0) + sig[:slow] = 0.0 + return sig + + +def oos(sig, df, frac=0.30): + split = int(len(df) * (1 - frac)) + s2 = sig.copy(); s2[:split] = 0.0 + return simulate_position(s2, df) + + +def show(label, df, sig): + full = simulate_position(sig, df) + o = oos(sig, df) + anni = f"{full['pos_years']}/{full['n_years']}" + print(f" {label:<26s}{full['flips']:>6d}{full['ret']:>+9.0f}{o['ret']:>+9.0f}" + f"{full['dd']:>6.0f}{full['exposure']:>6.0f}{anni:>8s}") + return full, o + + +if __name__ == "__main__": + assets = available_assets() + print(f"TSMOM / trend — fee {FEE_RT*100:.2f}% RT, leva3x pos15% | OOS30%") + for tf in ["1d", "4h"]: + print(f"\n ###### TF {tf} ######") + for a in assets: + df = get_df(a, tf) + print(f"\n === {a} {tf} === {'Flip':>5s}{'FULL%':>8s}{'OOS%':>8s}{'DD%':>6s}{'Exp%':>6s}{'AnniP':>8s}") + show("TSMOM lb30 long/short", df, tsmom_signal(df, 30)) + show("TSMOM lb30 long-only", df, tsmom_signal(df, 30, long_only=True)) + show("TSMOM lb90 long/short", df, tsmom_signal(df, 90)) + show("EMA 20/100 long/short", df, ema_dual_signal(df, 20, 100)) + show("EMA 20/100 long-only", df, ema_dual_signal(df, 20, 100, long_only=True)) diff --git a/scripts/strategies/DIP01_dip_reversion.py b/scripts/strategies/DIP01_dip_reversion.py new file mode 100644 index 0000000..a8d1e86 --- /dev/null +++ b/scripts/strategies/DIP01_dip_reversion.py @@ -0,0 +1,152 @@ +"""DIP01 — Dip-Buy Z-Score Reversion (long-only). + +Variante robusta e ONESTA della famiglia mean-reversion: compra SOLO i dip +(close a z<=-z_in deviazioni sotto la media mobile) e prende profitto al rientro +verso la media. Niente short: nel campione 2018-2026 shortare cripto perde OOS +sistematicamente (vedi scripts/analysis/honest_final.py). + +Logica: + 1. z-score = (close - SMA(n)) / STD(n) + 2. ENTRY long quando z attraversa al ribasso -z_in (capitolazione) + 3. EXIT: take-profit alla media mobile, stop-loss a sl_atr*ATR sotto l'entry, + o time-limit max_bars + 4. ingresso a close[i] (eseguibile dal vivo, nessun look-ahead) + +Validazione (netto, fee 0.10% RT Deribit, leva 3x, OOS = ultimo 30%): + BTC 1h: FULL +298% / OOS +59% / DD 23% / 7-9 anni positivi + ETH 1h: FULL +190% / OOS +224% / DD 54% + SOL 1h: FULL +50% / OOS +13% / DD 25% + Regge lo sweep fee fino a 0.20% RT (BTC OOS +45% anche a 0.20%). + Robusto su BTC/ETH/SOL (asset major); sugli alt molto parabolici (DOGE/BNB) + non ha edge -> usare solo su BTC/ETH/SOL. + +Compatibile con StrategyWorker: ogni Signal porta tp/sl/max_bars in metadata. +""" +from __future__ import annotations +import sys +sys.path.insert(0, ".") + +import numpy as np +import pandas as pd + +from src.strategies.base import Strategy, Signal, BacktestResult, YearlyStats, TF_MINUTES +from src.data.downloader import load_data + + +def _atr(df: pd.DataFrame, n: 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).rolling(n).mean().values + + +class DipReversion(Strategy): + name = "DIP01_dip_reversion" + description = "Long-only dip-buy z-score reversion, TP alla media" + default_assets = ["BTC", "ETH", "SOL"] + default_timeframes = ["1h"] + fee_rt = 0.001 + leverage = 3.0 + position_size = 0.15 + initial_capital = 1000.0 + + def generate_signals(self, df: pd.DataFrame, ts: pd.DatetimeIndex, + **params) -> list[Signal]: + c = df["close"].values + n = params.get("n", 50) + z_in = params.get("z_in", 2.5) + sl_atr = params.get("sl_atr", 2.5) + max_bars = params.get("max_bars", 24) + + ma = pd.Series(c).rolling(n).mean().values + sd = pd.Series(c).rolling(n).std().values + a = _atr(df, 14) + z = (c - ma) / np.where(sd == 0, np.nan, sd) + + signals: list[Signal] = [] + for i in range(n + 14, len(c)): + if np.isnan(z[i]) or np.isnan(a[i]): + continue + if z[i] <= -z_in and z[i - 1] > -z_in: + signals.append(Signal( + idx=i, direction=1, entry_price=c[i], + metadata={"tp": float(ma[i]), "sl": float(c[i] - sl_atr * a[i]), + "max_bars": max_bars}, + )) + return signals + + def backtest(self, asset: str, tf: str = "1h", hold: int = 3, + **params) -> BacktestResult | None: + df = load_data(asset, tf) + ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True) + signals = self.generate_signals(df, ts, **params) + if not signals: + return None + h, l, c = df["high"].values, df["low"].values, df["close"].values + n = len(c) + fee = self.fee_rt * self.leverage + capital = peak = float(self.initial_capital) + max_dd = 0.0 + total_bars = 0 + last_exit = -1 + yearly: dict[int, dict] = {} + + for sig in signals: + i, d = sig.idx, sig.direction + if i <= last_exit or i + 1 >= n: + continue + entry = c[i] + tp, sl, mb = sig.metadata["tp"], sig.metadata["sl"], sig.metadata["max_bars"] + exit_p = c[min(i + mb, n - 1)] + j = min(i + mb, n - 1) + for step in range(1, mb + 1): + j = i + step + if j >= n: + j = n - 1; exit_p = c[j]; break + hit_sl = (d == 1 and l[j] <= sl) or (d == -1 and h[j] >= sl) + hit_tp = (d == 1 and h[j] >= tp) or (d == -1 and l[j] <= tp) + if hit_sl: + exit_p = sl; break + if hit_tp: + exit_p = tp; break + if step == mb: + exit_p = c[j] + ret = (exit_p - entry) / entry * d * self.leverage - fee + capital = max(capital + capital * self.position_size * ret, 10.0) + if capital > peak: + peak = capital + max_dd = max(max_dd, (peak - capital) / peak) + total_bars += (j - i) + last_exit = j + year = ts.iloc[i].year + yr = yearly.setdefault(year, {"w": 0, "t": 0, "pnl": 0.0}) + yr["t"] += 1 + if ret > 0: + yr["w"] += 1 + yr["pnl"] += ret * self.initial_capital + + all_t = sum(v["t"] for v in yearly.values()) + all_w = sum(v["w"] for v in yearly.values()) + if all_t == 0: + return None + yearly_stats = [YearlyStats(y, v["t"], v["w"], v["pnl"]) for y, v in sorted(yearly.items())] + return BacktestResult( + strategy_name=self.name, asset=asset, timeframe=tf, params=params, + trades=all_t, wins=all_w, pnl=sum(v["pnl"] for v in yearly.values()), + capital=capital, initial_capital=self.initial_capital, + max_dd=max_dd * 100, time_in_market_pct=total_bars / n * 100, + avg_trade_duration_h=total_bars / all_t * TF_MINUTES.get(tf, 60) / 60, + years_active=len(yearly), yearly=yearly_stats, + ) + + +if __name__ == "__main__": + strat = DipReversion() + print(f"{'=' * 100}") + print(f" DIP01 DIP-BUY REVERSION — netto fee {strat.fee_rt*100:.2f}% RT, leva {strat.leverage:.0f}x") + print(f"{'=' * 100}") + for asset in ["BTC", "ETH", "SOL"]: + r = strat.backtest(asset, "1h", n=50, z_in=2.5, sl_atr=2.5, max_bars=24) + if r: + r.strategy_name = f"DIP01 {asset} 1h" + r.print_summary() diff --git a/scripts/strategies/ROT01_xsect_rotation.py b/scripts/strategies/ROT01_xsect_rotation.py new file mode 100644 index 0000000..39265b0 --- /dev/null +++ b/scripts/strategies/ROT01_xsect_rotation.py @@ -0,0 +1,48 @@ +"""ROT01 — Cross-Sectional Momentum Rotation (multi-crypto, long-only), 1d. + +UNA strategia che usa l'INTERO paniere di crypto in un solo book: ogni giorno +ordina gli asset per momentum (rendimento sugli ultimi `lookback` giorni) e alloca +il capitale in parti uguali ai `top_k` con momentum positivo; il resto in cash. +Cattura la dispersione tra crypto (gli alt forti corrono molto piu' di BTC nei bull) +senza shortare nulla. Meccanismo distinto da DIP01/TR01 -> vera diversificazione. + +Onesto: i pesi a close[i] usano solo rendimenti passati; il rendimento del giorno +i->i+1 e' realizzato con quei pesi. Fee sul turnover. Allineamento per timestamp. + +Validazione (netto, fee 0.10% RT, gross 0.45, OOS = ultimo 30%): + lb=60 top2 -> FULL +679% / OOS +44% / DD 53% / 5-7 anni positivi. + Param-insensitive (tutte le lb/k positive) e regge fee fino 0.20% RT (OOS +41%). + Per-anno: 2020+33 2021+181 2022-29 2023+43 2024+59 2025+6 2026-10 (i negativi = bear). +Dettagli in scripts/analysis/honest_rotation.py / honest_final.py. +""" +from __future__ import annotations + +import sys +from pathlib import Path + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(PROJECT_ROOT)) + +from scripts.analysis.honest_rotation import build_panel, simulate_rotation # noqa: E402 +from scripts.analysis.honest_lab import available_assets + +LOOKBACK, TOP_K, TF = 60, 2, "1d" + + +def run(): + assets = available_assets() + panel = build_panel(assets, TF) + print("=" * 90) + print(f" ROT01 ROTAZIONE cross-sectional momentum | {TF} lb={LOOKBACK} top{TOP_K} | netto fee 0.10% RT") + print("=" * 90) + print(f" Paniere: {list(panel.columns)}") + print(f" Periodo: {panel.index[0].date()} -> {panel.index[-1].date()} ({panel.shape[0]} barre)") + full = simulate_rotation(panel, lookback=LOOKBACK, top_k=TOP_K, fee_rt=0.001) + oos = simulate_rotation(panel, lookback=LOOKBACK, top_k=TOP_K, fee_rt=0.001, oos_frac=0.30) + print(f"\n FULL: {full['ret']:+.0f}% DD {full['dd']:.0f}% turnover {full['turnover']:.0f}") + print(f" OOS : {oos['ret']:+.0f}% DD {oos['dd']:.0f}% ({full['pos_years']}/{full['n_years']} anni positivi)") + print(" Per-anno: " + " ".join(f"{y}:{v:+.0f}%" for y, v in sorted(full["yearly"].items()))) + + +if __name__ == "__main__": + run() diff --git a/scripts/strategies/TR01_ema_trend.py b/scripts/strategies/TR01_ema_trend.py new file mode 100644 index 0000000..22ed671 --- /dev/null +++ b/scripts/strategies/TR01_ema_trend.py @@ -0,0 +1,50 @@ +"""TR01 — EMA Trend Following (long-only), timeframe 4h. + +Cavalca i trend rialzisti, si mette in cash nei downtrend. Niente short +(shortare cripto perde OOS nel campione 2018-2026). Complementare a DIP01: +DIP01 guadagna nei regimi di reversione, TR01 nei regimi di trend. + +Logica: + 1. EMA fast (20) e EMA slow (100) sul close + 2. LONG quando EMA_fast > EMA_slow (uptrend), altrimenti CASH + 3. posizione continua, decisione a close[i] (no look-ahead); + fee solo sui cambi di stato (poche operazioni = fee non letali) + +Validazione (netto, fee 0.10% RT, leva 3x, pos 15%, OOS = ultimo 30%): + robusto FULL+OOS su 5/8 asset: BNB(+14), BTC(+27), DOGE(+53), SOL(+7), XRP(+29) OOS. + ETH ~flat, ADA/LTC negativi OOS -> preferire BNB/BTC/DOGE/SOL/XRP. +Dettagli in scripts/analysis/honest_final.py / honest_trend.py. +""" +from __future__ import annotations + +import sys +from pathlib import Path + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(PROJECT_ROOT)) + +from scripts.analysis.honest_trend import ( # noqa: E402 + simulate_position, ema_dual_signal, oos as trend_oos, +) +from scripts.analysis.honest_lab import get_df + +ASSETS = ["BNB", "BTC", "DOGE", "SOL", "XRP"] +FAST, SLOW, TF = 20, 100, "4h" + + +def run(): + print("=" * 90) + print(f" TR01 EMA TREND {FAST}/{SLOW} long-only | {TF} | netto fee 0.10% RT leva 3x pos 15%") + print("=" * 90) + print(f" {'Asset':<6s}{'Flip':>6s}{'FULL%':>9s}{'OOS%':>9s}{'DD%':>6s}{'Exp%':>6s}{'AnniPos':>9s}") + for a in ASSETS: + df = get_df(a, TF) + sig = ema_dual_signal(df, FAST, SLOW, long_only=True) + f = simulate_position(sig, df) + o = trend_oos(sig, df) + print(f" {a:<6s}{f['flips']:>6d}{f['ret']:>+9.0f}{o['ret']:>+9.0f}" + f"{f['dd']:>6.0f}{f['exposure']:>6.0f}{str(f['pos_years'])+'/'+str(f['n_years']):>9s}") + + +if __name__ == "__main__": + run()