"""Migliorare Acc e ridurre DD sulle fade (MR01/MR02/MR03/MR07) senza overfit. Leve testate, ognuna misurata FULL e OOS (ultimo 30%) per non illudersi: - vol-target sizing: size per trade ~ 1/distanza-SL -> rischio costante, DD piu' liscio - filtro vol regime: salta i trade con ATR% in coda alta (periodi caotici) - filtro anti-trend: non fadare contro un trend forte (|close-EMA_long|/ATR grande) - portfolio: equity curve combinata delle 4 strategie su un conto unico Engine fedele (ingresso close[i], exit TP/SL intrabar o time-limit, non-overlap, capitale composto) con sizing per-trade. Numeri NETTI fee 0.10% RT, leva 3x. """ 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 src.data.downloader import load_data from scripts.analysis.strategy_research import bollinger_fade, atr from scripts.analysis.strategy_research_v2 import donchian_fade, keltner_fade, return_reversal FEE_RT, LEV, POS, INIT, OOS_FRAC = 0.001, 3.0, 0.15, 1000.0, 0.30 # config base di ogni strategia (come strategies.yml) STRATS = { "MR01": (bollinger_fade, dict(n=50, k=2.5, sl_atr=2.0, max_bars=24)), "MR02": (donchian_fade, dict(n=20, sl_atr=2.0, max_bars=24)), "MR03": (keltner_fade, dict(n=30, k=2.0, sl_atr=2.0, max_bars=24)), "MR07": (return_reversal,dict(n=50, k=3.5, tp_atr=2.0, sl_atr=1.5, max_bars=24)), } STRATS_ETH3 = dict(STRATS); STRATS_ETH3["MR03"] = (keltner_fade, dict(n=50, k=2.0, sl_atr=2.0, max_bars=24)) def add_context(ents, df, ema_long=200): """Aggiunge a ogni entry: sl_dist_pct, atr_pct, trend_dist (|close-EMA|/ATR).""" c = df["close"].values a = atr(df, 14) el = pd.Series(c).ewm(span=ema_long, adjust=False).mean().values apct = a / c for e in ents: i = e["i"] e["sl_dist"] = abs(c[i] - e["sl"]) / c[i] e["atr_pct"] = apct[i] e["trend_dist"] = abs(c[i] - el[i]) / a[i] if a[i] else 0.0 return ents def simulate(ents, df, fee_rt=FEE_RT, lev=LEV, split=-1, sizer=None, vol_skip=None, trend_skip=None, max_size=0.30): """sizer: funzione(entry)->frazione capitale; default POS fisso. vol_skip: soglia atr_pct sopra cui salto. trend_skip: soglia trend_dist sopra cui salto.""" h, l, c = df["high"].values, df["low"].values, df["close"].values n = len(c) ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True) cap = peak = INIT dd = 0.0; last = -1; trd = wins = 0 fee = fee_rt * lev yearly = {}; rets = [] for e in ents: i, d = e["i"], e["d"] if i <= last or i + 1 >= n or i < split: continue if vol_skip is not None and e["atr_pct"] > vol_skip: continue if trend_skip is not None and e["trend_dist"] > trend_skip: continue entry = c[i]; tp, sl, mb = e["tp"], e["sl"], e["max_bars"] exit_p = c[min(i + mb, n - 1)]; j = min(i + mb, n - 1) for k in range(1, mb + 1): j = i + k if j >= n: exit_p = c[n - 1]; break hs = (d == 1 and l[j] <= sl) or (d == -1 and h[j] >= sl) ht = (d == 1 and h[j] >= tp) or (d == -1 and l[j] <= tp) if hs: exit_p = sl; break if ht: exit_p = tp; break if k == mb: exit_p = c[j] ret = (exit_p - entry) / entry * d * lev - fee size = POS if sizer is None else min(sizer(e), max_size) cap = max(cap + cap * size * ret, 10.0) peak = max(peak, cap); dd = max(dd, (peak - cap) / peak) trd += 1; wins += ret > 0; last = j; rets.append(ret * size) y = ts.iloc[i].year; yearly[y] = yearly.get(y, 0.0) + ret * size * INIT sharpe = float(np.mean(rets) / np.std(rets) * np.sqrt(len(rets))) if len(rets) > 1 and np.std(rets) > 0 else 0.0 return dict(trades=trd, acc=wins / trd * 100 if trd else 0.0, ret=(cap / INIT - 1) * 100, dd=dd * 100, yearly=yearly, sharpe=sharpe) def vol_target_sizer(target=0.015): """size t.c. rischio (size*lev*sl_dist) ~ target; piu' largo lo stop, meno size.""" return lambda e: target / (LEV * max(e["sl_dist"], 1e-4)) def line(label, full, oos): print(f" {label:<28s}{full['trades']:>6d}{full['acc']:>7.1f}{full['ret']:>+10.0f}{full['dd']:>7.1f}{full['sharpe']:>7.2f}" f" | {oos['trades']:>5d}{oos['acc']:>7.1f}{oos['ret']:>+9.0f}{oos['dd']:>7.1f}{oos['sharpe']:>7.2f}") def main(): for asset in ["BTC", "ETH"]: df = load_data(asset, "1h") split = int(len(df) * (1 - OOS_FRAC)) table = STRATS_ETH3 if asset == "ETH" else STRATS # quantili vol globali per la soglia (p90) print("\n" + "=" * 110) print(f" {asset} 1h — leve di riduzione DD / aumento Acc | NETTO fee 0.10% RT, leva 3x") print("=" * 110) print(f" {'config':<28s}{'Trd':>6s}{'Acc%':>7s}{'Ret%':>10s}{'DD%':>7s}{'Shrp':>7s}" f" | {'oTrd':>5s}{'oAcc':>7s}{'oRet':>9s}{'oDD':>7s}{'oShrp':>7s}") print(" " + "-" * 106) for nm, (fn, params) in table.items(): ents = add_context(fn(df, **params), df) apct = np.array([e["atr_pct"] for e in ents]) p85 = float(np.quantile(apct, 0.85)) tdist = np.array([e["trend_dist"] for e in ents]) t90 = float(np.quantile(tdist, 0.90)) base_f = simulate(ents, df); base_o = simulate(ents, df, split=split) line(f"{nm} base", base_f, base_o) vt_f = simulate(ents, df, sizer=vol_target_sizer()); vt_o = simulate(ents, df, split=split, sizer=vol_target_sizer()) line(f"{nm} +volTarget", vt_f, vt_o) vs_f = simulate(ents, df, vol_skip=p85); vs_o = simulate(ents, df, split=split, vol_skip=p85) line(f"{nm} +volSkip(p85)", vs_f, vs_o) ts_f = simulate(ents, df, trend_skip=t90); ts_o = simulate(ents, df, split=split, trend_skip=t90) line(f"{nm} +trendSkip(p90)", ts_f, ts_o) allf = simulate(ents, df, sizer=vol_target_sizer(), vol_skip=p85, trend_skip=t90) allo = simulate(ents, df, split=split, sizer=vol_target_sizer(), vol_skip=p85, trend_skip=t90) line(f"{nm} +ALL", allf, allo) print(" " + "-" * 106) print("\n Shrp = Sharpe annuo-naive sui ritorni per-trade. oXxx = stessa metrica su OOS (ultimo 30%).") if __name__ == "__main__": main()