feat(analysis): 3 strategie oneste validate OOS multi-crypto (DIP/TR/ROT)
Ricerca onesta post-squeeze su 8 crypto (2018-2026), engine fee-aware con ingresso eseguibile a close[i], uscita TP/SL intrabar, OOS held-out, sweep fee. Lezione madre: shortare cripto perde OOS sistematicamente (campione net-bull) -> tutte le strategie robuste sono long-biased. Tre meccanismi distinti e complementari: - DIP01 dip-buy z-score reversion (long-only, 1h) robusto BTC/ETH/SOL - TR01 EMA 20/100 trend-following (long-only, 4h) robusto su 5/8 asset - ROT01 rotazione cross-sectional momentum sul paniere (1d) OOS +44%, param-insensitive Engine e validazione: scripts/analysis/honest_lab.py + honest_final.py (+ honest_candidates/diag/diag2/trend/rotation). Diario in docs/diary/. Onesto sull'obiettivo: €50/giorno su €1000 in pochi mesi non e' raggiungibile a rischio sano (~1825%/anno); edge reali 30-60% OOS pluriennale. Via realistica: portafoglio delle 3, leva moderata, crescita composta. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -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()
|
||||
@@ -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()
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user