merge: strategy_free -> main (fade MR02/MR03/MR07 + filtro trend Acc/DD)
Integra il lavoro su branch strategy_free, indipendente dalle strategie oneste (DIP/TR/ROT/PORT) gia' su main: nessun file in comune, merge pulito. - 3 nuove fade mean-reversion validate OOS fee-aware: MR02 Donchian fade, MR03 Keltner fade, MR07 Return reversal (+ base condivisa fade_base). - Filtro trend (trend_max/ema_long) su tutte le fade: alza Acc e riduce DD (drastico su ETH), edge OOS confermato; modello portafoglio equipesato.
This commit is contained in:
@@ -0,0 +1,139 @@
|
||||
"""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()
|
||||
@@ -0,0 +1,163 @@
|
||||
"""Affina il filtro trend (soglia assoluta ATR) e costruisce il portafoglio combinato.
|
||||
|
||||
Due risultati:
|
||||
(1) trend filter: salta le fade quando |close-EMA200|/ATR > soglia (non fadare un
|
||||
trend estremo). Soglia ASSOLUTA in multipli di ATR -> stessa regola per tutte
|
||||
le strategie/asset, basso rischio di overfit. Sweep soglie, FULL e OOS.
|
||||
(2) portafoglio: equity curve combinata delle 4 strategie sullo stesso conto
|
||||
(rischio diviso fra N posizioni). Curve poco correlate -> DD aggregato << DD
|
||||
della singola strategia. Confronto singola vs portafoglio, con/senza filtro.
|
||||
"""
|
||||
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, INIT, OOS_FRAC = 0.001, 3.0, 1000.0, 0.30
|
||||
|
||||
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_ETH = dict(STRATS); STRATS_ETH["MR03"] = (keltner_fade, dict(n=50, k=2.0, sl_atr=2.0, max_bars=24))
|
||||
|
||||
|
||||
def build_trades(ents, df, lev=LEV, fee_rt=FEE_RT, trend_max=None, ema_long=200):
|
||||
"""Ritorna lista trade non-overlap: (entry_idx, exit_idx, ret_netto). Filtro trend opzionale."""
|
||||
h, l, c = df["high"].values, df["low"].values, df["close"].values
|
||||
n = len(c); a = atr(df, 14)
|
||||
el = pd.Series(c).ewm(span=ema_long, adjust=False).mean().values
|
||||
fee = fee_rt * lev
|
||||
out = []; last = -1
|
||||
for e in ents:
|
||||
i, d = e["i"], e["d"]
|
||||
if i <= last or i + 1 >= n:
|
||||
continue
|
||||
if trend_max is not None and a[i] and abs(c[i] - el[i]) / a[i] > trend_max:
|
||||
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
|
||||
out.append((i, j, ret)); last = j
|
||||
return out
|
||||
|
||||
|
||||
def metrics_single(trades, ts, pos=0.15, split=-1):
|
||||
cap = peak = INIT; dd = 0.0; trd = wins = 0; rets = []
|
||||
for i, j, ret in trades:
|
||||
if i < split:
|
||||
continue
|
||||
cap = max(cap + cap * pos * ret, 10.0)
|
||||
peak = max(peak, cap); dd = max(dd, (peak - cap) / peak)
|
||||
trd += 1; wins += ret > 0; rets.append(ret * pos)
|
||||
sh = 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, sharpe=sh)
|
||||
|
||||
|
||||
def sleeve_equity(trades, n_bars, pos=0.15, split=-1):
|
||||
"""Equity curve di uno sleeve su sotto-conto indipendente (capitale INIT, pos fissa).
|
||||
Ritorna array lungo n_bars (step aggiornato alla chiusura di ogni trade)."""
|
||||
eq = np.full(n_bars, INIT, dtype=float)
|
||||
cap = INIT
|
||||
for i, j, ret in sorted(trades, key=lambda t: t[1]):
|
||||
if i < split:
|
||||
continue
|
||||
cap = max(cap + cap * pos * ret, 10.0)
|
||||
eq[j:] = cap # da j in poi il sotto-conto vale cap
|
||||
return eq
|
||||
|
||||
|
||||
def metrics_portfolio(strat_trades, n_bars, ts, pos=0.15, split=-1):
|
||||
"""Portafoglio equipesato: capitale diviso in N sotto-conti indipendenti, ciascuno
|
||||
con la sua strategia a `pos` fisso. Equity aggregata = media dei sotto-conti (somma
|
||||
normalizzata a base INIT). DD misurato sull'equity aggregata. Niente leva sovrapposta."""
|
||||
sleeves = [sleeve_equity(tr, n_bars, pos=pos, split=split) for tr in strat_trades.values()]
|
||||
agg = np.mean(sleeves, axis=0) # media -> base INIT, diversificazione reale
|
||||
# restringi alla finestra effettiva (da split in poi se OOS)
|
||||
lo = max(split, 0)
|
||||
agg = agg[lo:]
|
||||
peak = np.maximum.accumulate(agg)
|
||||
dd = float(np.max((peak - agg) / peak) * 100)
|
||||
trd = sum(1 for tr in strat_trades.values() for i, _, _ in tr if i >= split)
|
||||
wins = sum(1 for tr in strat_trades.values() for i, _, r in tr if i >= split and r > 0)
|
||||
return dict(trades=trd, acc=wins / trd * 100 if trd else 0.0,
|
||||
ret=(agg[-1] / INIT - 1) * 100, dd=dd, sharpe=0.0)
|
||||
|
||||
|
||||
def main():
|
||||
# ---------- (1) sweep soglia trend ----------
|
||||
print("=" * 104)
|
||||
print(" (1) FILTRO TREND |close-EMA200|/ATR > soglia -> SALTA | NETTO fee 0.10% RT, leva 3x")
|
||||
print("=" * 104)
|
||||
print(f" {'Strat/Asset':<14s}{'soglia':>8s}{'Trd':>6s}{'Acc%':>7s}{'Ret%':>9s}{'DD%':>7s}"
|
||||
f" | {'oAcc':>6s}{'oRet':>9s}{'oDD':>7s}{'oShrp':>7s}")
|
||||
print(" " + "-" * 100)
|
||||
thresholds = [None, 4.0, 3.0, 2.5, 2.0]
|
||||
for asset in ["BTC", "ETH"]:
|
||||
df = load_data(asset, "1h"); ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
|
||||
split = int(len(df) * (1 - OOS_FRAC))
|
||||
table = STRATS_ETH if asset == "ETH" else STRATS
|
||||
for nm, (fn, params) in table.items():
|
||||
ents = fn(df, **params)
|
||||
for thr in thresholds:
|
||||
tr = build_trades(ents, df, trend_max=thr)
|
||||
f = metrics_single(tr, ts); o = metrics_single(tr, ts, split=split)
|
||||
lab = "base" if thr is None else f"{thr}ATR"
|
||||
print(f" {nm+' '+asset:<14s}{lab:>8s}{f['trades']:>6d}{f['acc']:>7.1f}{f['ret']:>+9.0f}{f['dd']:>7.1f}"
|
||||
f" | {o['acc']:>6.1f}{o['ret']:>+9.0f}{o['dd']:>7.1f}{o['sharpe']:>7.2f}")
|
||||
print(" " + "-" * 100)
|
||||
|
||||
# ---------- (2) portafoglio combinato ----------
|
||||
print("\n" + "=" * 104)
|
||||
print(" (2) PORTAFOGLIO equipesato: capitale diviso in N sotto-conti indipendenti")
|
||||
print(" (pos 0.15 ciascuno, filtro trend 3.0 ATR). DD aggregato vs singola strategia.")
|
||||
print("=" * 104)
|
||||
print(f" {'Universo':<26s}{'Trd':>6s}{'Acc%':>7s}{'Ret%':>10s}{'DD%':>7s}{'':>7s}"
|
||||
f" | {'oAcc':>6s}{'oRet':>9s}{'oDD':>7s}{'':>7s}")
|
||||
print(" " + "-" * 100)
|
||||
all_trades = {}
|
||||
for asset in ["BTC", "ETH"]:
|
||||
df = load_data(asset, "1h"); ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
|
||||
split = int(len(df) * (1 - OOS_FRAC)); n = len(df)
|
||||
table = STRATS_ETH if asset == "ETH" else STRATS
|
||||
st = {f"{nm}_{asset}": build_trades(fn(df, **p), df, trend_max=3.0) for nm, (fn, p) in table.items()}
|
||||
all_trades.update(st)
|
||||
f = metrics_portfolio(st, n, ts); o = metrics_portfolio(st, n, ts, split=split)
|
||||
print(f" {'Portafoglio '+asset+' (4 strat)':<26s}{f['trades']:>6d}{f['acc']:>7.1f}{f['ret']:>+10.0f}{f['dd']:>7.1f}{f['sharpe']:>7.2f}"
|
||||
f" | {o['acc']:>6.1f}{o['ret']:>+9.0f}{o['dd']:>7.1f}{o['sharpe']:>7.2f}")
|
||||
# globale 8 sleeve
|
||||
df0 = load_data("BTC", "1h"); ts0 = pd.to_datetime(df0["timestamp"], unit="ms", utc=True)
|
||||
split0 = int(len(df0) * (1 - OOS_FRAC))
|
||||
f = metrics_portfolio(all_trades, len(df0), ts0); o = metrics_portfolio(all_trades, len(df0), ts0, split=split0)
|
||||
print(" " + "-" * 100)
|
||||
print(f" {'GLOBALE BTC+ETH (8 sleeve)':<26s}{f['trades']:>6d}{f['acc']:>7.1f}{f['ret']:>+10.0f}{f['dd']:>7.1f}{f['sharpe']:>7.2f}"
|
||||
f" | {o['acc']:>6.1f}{o['ret']:>+9.0f}{o['dd']:>7.1f}{o['sharpe']:>7.2f}")
|
||||
print("\n Nota: ogni sleeve gira su un sotto-conto indipendente (pos 0.15); l'equity di")
|
||||
print(" portafoglio e' la media dei sotto-conti. Curve poco correlate => DD aggregato")
|
||||
print(" molto piu' basso del DD del singolo sleeve (la vera leva anti-drawdown).")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,306 @@
|
||||
"""Ricerca v2 — nuove strategie oltre MR01, stessa metodologia fee-aware OOS.
|
||||
|
||||
Lezioni ereditate (vedi strategy_research.py / oos_validation.py):
|
||||
- mean-reversion ha edge, continuation/trend NO (i breakout rientrano)
|
||||
- fee = vincolo di prim'ordine -> default Deribit 0.10% RT, poche operazioni meglio
|
||||
- ingresso ESEGUIBILE a close[i] (mai look-ahead con direzione da barra i)
|
||||
- ogni numero NETTO dopo fee+leva, su finestra held-out (OOS=ultimo 30%) + per anno
|
||||
|
||||
Nuovi candidati (tutti fade/mean-reversion con ingresso onesto):
|
||||
MR02 donchian_fade - fade rottura canale Donchian (opposto del trend che muore)
|
||||
MR03 keltner_fade - fade canale Keltner (ATR), TP alla EMA media
|
||||
MR04 zscore_revert - fade deviazione z-score estrema, TP alla media
|
||||
MR05 boll_fade_adx - Bollinger fade con filtro regime ADX (solo mercato laterale)
|
||||
|
||||
Engine identico a strategy_research.simulate (ingresso close[i], exit TP/SL intrabar
|
||||
high/low o time-limit, non-overlap, capitale composto).
|
||||
"""
|
||||
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))
|
||||
|
||||
# riusa engine, dati e indicatori gia' validati
|
||||
from scripts.analysis.strategy_research import (
|
||||
FEE_RT, LEV, POS, OOS_FRAC, get_df, atr, rsi, simulate,
|
||||
)
|
||||
|
||||
|
||||
# --------------------------- indicatori extra ---------------------------
|
||||
def ema(x: np.ndarray, n: int) -> np.ndarray:
|
||||
return pd.Series(x).ewm(span=n, adjust=False).mean().values
|
||||
|
||||
|
||||
def adx(df: pd.DataFrame, n: int = 14) -> np.ndarray:
|
||||
"""Average Directional Index: misura la forza del trend (alto=trend, basso=range)."""
|
||||
h, l, c = df["high"].values, df["low"].values, df["close"].values
|
||||
up = h - np.roll(h, 1)
|
||||
dn = np.roll(l, 1) - l
|
||||
up[0] = dn[0] = 0.0
|
||||
plus_dm = np.where((up > dn) & (up > 0), up, 0.0)
|
||||
minus_dm = np.where((dn > up) & (dn > 0), dn, 0.0)
|
||||
pc = np.roll(c, 1); pc[0] = c[0]
|
||||
tr = np.maximum(h - l, np.maximum(np.abs(h - pc), np.abs(l - pc)))
|
||||
atr_n = pd.Series(tr).ewm(alpha=1/n, adjust=False).mean().values
|
||||
pdi = 100 * pd.Series(plus_dm).ewm(alpha=1/n, adjust=False).mean().values / np.where(atr_n == 0, np.nan, atr_n)
|
||||
mdi = 100 * pd.Series(minus_dm).ewm(alpha=1/n, adjust=False).mean().values / np.where(atr_n == 0, np.nan, atr_n)
|
||||
dx = 100 * np.abs(pdi - mdi) / np.where((pdi + mdi) == 0, np.nan, pdi + mdi)
|
||||
return pd.Series(dx).ewm(alpha=1/n, adjust=False).mean().values
|
||||
|
||||
|
||||
# --------------------------- strategie nuove ---------------------------
|
||||
def donchian_fade(df, n=20, sl_atr=2.0, max_bars=24):
|
||||
"""MR02 — fade rottura canale Donchian: rompe sopra max-N => short verso il mid.
|
||||
|
||||
Coerente con 'i breakout rientrano': l'opposto di donchian_trend (che fallisce).
|
||||
Ingresso a close[i] sulla barra che chiude oltre il canale precedente.
|
||||
TP al centro del canale, SL = sl_atr*ATR oltre l'estremo.
|
||||
"""
|
||||
h, l, c = df["high"].values, df["low"].values, df["close"].values
|
||||
hh = pd.Series(h).rolling(n).max().shift(1).values
|
||||
ll = pd.Series(l).rolling(n).min().shift(1).values
|
||||
a = atr(df, 14)
|
||||
ents = []
|
||||
for i in range(n + 14, len(c)):
|
||||
if np.isnan(hh[i]) or np.isnan(a[i]):
|
||||
continue
|
||||
mid = (hh[i] + ll[i]) / 2.0
|
||||
if c[i] > hh[i] and c[i - 1] <= hh[i - 1]: # rottura rialzista => fade short
|
||||
ents.append({"i": i, "d": -1, "tp": mid, "sl": c[i] + sl_atr * a[i], "max_bars": max_bars})
|
||||
elif c[i] < ll[i] and c[i - 1] >= ll[i - 1]: # rottura ribassista => fade long
|
||||
ents.append({"i": i, "d": 1, "tp": mid, "sl": c[i] - sl_atr * a[i], "max_bars": max_bars})
|
||||
return ents
|
||||
|
||||
|
||||
def keltner_fade(df, n=20, k=2.0, sl_atr=2.0, max_bars=24):
|
||||
"""MR03 — fade canale Keltner (EMA +/- k*ATR), TP alla EMA media.
|
||||
|
||||
Come Bollinger ma banda basata su ATR (volatilita' di range) invece che std:
|
||||
reagisce diversamente ai gap. Ingresso quando close esce dalla banda.
|
||||
"""
|
||||
c = df["close"].values
|
||||
e = ema(c, n)
|
||||
a = atr(df, n)
|
||||
up, lo = e + k * a, e - k * a
|
||||
ents = []
|
||||
for i in range(n + 1, 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": e[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": e[i], "sl": c[i] + sl_atr * a[i], "max_bars": max_bars})
|
||||
return ents
|
||||
|
||||
|
||||
def zscore_revert(df, n=50, z=2.0, sl_atr=2.5, max_bars=24):
|
||||
"""MR04 — fade deviazione z-score estrema dalla media, TP alla media.
|
||||
|
||||
z = (close-ma)/std. Entra quando |z| supera la soglia (close fuori); chiude
|
||||
quando torna alla media. Banda di Bollinger riparametrizzata in z (equivalente
|
||||
a k=z) ma con SL piu' largo e finestra lunga: poche operazioni, alta selettivita'.
|
||||
"""
|
||||
c = df["close"].values
|
||||
ma = pd.Series(c).rolling(n).mean().values
|
||||
sd = pd.Series(c).rolling(n).std().values
|
||||
a = atr(df, 14)
|
||||
ents = []
|
||||
for i in range(n + 14, len(c)):
|
||||
if np.isnan(ma[i]) or sd[i] == 0 or np.isnan(a[i]):
|
||||
continue
|
||||
zi = (c[i] - ma[i]) / sd[i]
|
||||
zp = (c[i - 1] - ma[i - 1]) / sd[i - 1] if sd[i - 1] else 0.0
|
||||
if zi <= -z and zp > -z:
|
||||
ents.append({"i": i, "d": 1, "tp": ma[i], "sl": c[i] - sl_atr * a[i], "max_bars": max_bars})
|
||||
elif zi >= z and zp < z:
|
||||
ents.append({"i": i, "d": -1, "tp": ma[i], "sl": c[i] + sl_atr * a[i], "max_bars": max_bars})
|
||||
return ents
|
||||
|
||||
|
||||
def boll_fade_adx(df, n=50, k=2.5, sl_atr=2.0, max_bars=24, adx_max=25.0):
|
||||
"""MR05 — Bollinger fade SOLO in regime laterale (ADX < adx_max).
|
||||
|
||||
Il fade soffre quando c'e' trend forte (il prezzo continua oltre la banda).
|
||||
Filtro ADX: opera solo quando la forza del trend e' bassa -> meno trade, edge piu' pulito.
|
||||
"""
|
||||
c = df["close"].values
|
||||
ma = pd.Series(c).rolling(n).mean().values
|
||||
sd = pd.Series(c).rolling(n).std().values
|
||||
a = atr(df, 14)
|
||||
ax = adx(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]) or np.isnan(ax[i]):
|
||||
continue
|
||||
if ax[i] >= adx_max: # trend forte: niente fade
|
||||
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 rsi2_fade(df, rsi_n=2, lo=10, hi=90, ma_n=200, tp_atr=2.0, sl_atr=3.0, max_bars=24):
|
||||
"""MR06 — Connors RSI(2) pullback in direzione del trend, TP/SL in ATR.
|
||||
|
||||
Meccanismo distinto da MR01/MR03: non usa bande di prezzo ma l'oscillatore
|
||||
RSI(2), che satura su micro-estremi. Filtro di trend con SMA lunga:
|
||||
- close SOPRA la SMA (uptrend) + RSI(2) < lo (dip) -> long, target rimbalzo
|
||||
- close SOTTO la SMA (downtrend) + RSI(2) > hi (pop) -> short
|
||||
TP = tp_atr*ATR a favore, SL = sl_atr*ATR contro. Compra il ritracciamento
|
||||
nel trend, non il contro-trend.
|
||||
"""
|
||||
c = df["close"].values
|
||||
r = rsi(c, rsi_n)
|
||||
ma = pd.Series(c).rolling(ma_n).mean().values
|
||||
a = atr(df, 14)
|
||||
ents = []
|
||||
for i in range(ma_n + 14, len(c)):
|
||||
if np.isnan(r[i]) or np.isnan(ma[i]) or np.isnan(a[i]):
|
||||
continue
|
||||
if r[i] < lo and c[i] > ma[i]: # dip in uptrend -> long
|
||||
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 r[i] > hi and c[i] < ma[i]: # pop in downtrend -> short
|
||||
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
|
||||
|
||||
|
||||
def return_reversal(df, n=50, k=3.5, tp_atr=2.0, sl_atr=1.5, max_bars=24):
|
||||
"""MR07 — fade movimento di barra estremo (return reversal).
|
||||
|
||||
Misura il rendimento dell'ultima barra in unita' di deviazione standard rolling
|
||||
dei rendimenti. Se |ret| > k*sigma, fada nella direzione opposta; TP/SL in ATR.
|
||||
Meccanismo distinto: usa la volatilita' dei RENDIMENTI, non i livelli di prezzo.
|
||||
Config robusta (k=3.5, tp=2ATR, sl=1.5ATR): positivo full+OOS BTC e ETH 1h,
|
||||
DD piu' contenuto (BTC 25% / ETH 46%).
|
||||
"""
|
||||
c = df["close"].values
|
||||
ret = np.zeros_like(c)
|
||||
ret[1:] = np.diff(c) / c[:-1]
|
||||
sig = pd.Series(ret).rolling(n).std().values
|
||||
a = atr(df, 14)
|
||||
ents = []
|
||||
for i in range(n + 14, len(c)):
|
||||
if np.isnan(sig[i]) or sig[i] == 0 or np.isnan(a[i]):
|
||||
continue
|
||||
z = ret[i] / sig[i]
|
||||
if z <= -k: # crollo di barra -> fade long
|
||||
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 z >= k: # spike di barra -> fade short
|
||||
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
|
||||
|
||||
|
||||
CANDIDATES = {
|
||||
"MR02 donch_fade n20": (donchian_fade, dict(n=20, sl_atr=2.0, max_bars=24)),
|
||||
"MR02 donch_fade n50": (donchian_fade, dict(n=50, sl_atr=2.0, max_bars=24)),
|
||||
"MR03 kelt_fade k2": (keltner_fade, dict(n=20, k=2.0, sl_atr=2.0, max_bars=24)),
|
||||
"MR03 kelt_fade k2.5": (keltner_fade, dict(n=20, k=2.5, sl_atr=2.0, max_bars=24)),
|
||||
"MR04 zscore z2 n50": (zscore_revert, dict(n=50, z=2.0, sl_atr=2.5, max_bars=24)),
|
||||
"MR04 zscore z2.5 n50": (zscore_revert, dict(n=50, z=2.5, sl_atr=2.5, max_bars=24)),
|
||||
"MR05 boll_adx n50": (boll_fade_adx, dict(n=50, k=2.5, sl_atr=2.0, max_bars=24, adx_max=25)),
|
||||
"MR05 boll_adx n20": (boll_fade_adx, dict(n=20, k=2.5, sl_atr=2.0, max_bars=24, adx_max=25)),
|
||||
"MR06 rsi2 10/90": (rsi2_fade, dict(rsi_n=2, lo=10, hi=90, ma_n=200, tp_atr=2.0, sl_atr=3.0, max_bars=24)),
|
||||
"MR06 rsi2 5/95": (rsi2_fade, dict(rsi_n=2, lo=5, hi=95, ma_n=200, tp_atr=2.0, sl_atr=3.0, max_bars=24)),
|
||||
"MR07 retrev k3.5": (return_reversal, dict(n=50, k=3.5, tp_atr=2.0, sl_atr=1.5, max_bars=24)),
|
||||
"MR07 retrev k3.0": (return_reversal, dict(n=50, k=3.0, tp_atr=2.0, sl_atr=1.5, max_bars=24)),
|
||||
}
|
||||
|
||||
|
||||
def table():
|
||||
print("=" * 122)
|
||||
print(f" RICERCA v2 — NETTO dopo fee {FEE_RT*100:.2f}% RT | leva {LEV:.0f}x | pos {POS*100:.0f}% "
|
||||
f"| OOS = ultimo {int(OOS_FRAC*100)}%")
|
||||
print("=" * 122)
|
||||
print(f" {'Strategia':<22s}{'Asset':>5s}{'TF':>5s}{'Trd':>6s}{'Tr/yr':>7s}{'Win%':>7s}"
|
||||
f"{'Ret%FULL':>10s}{'Ret%OOS':>10s}{'DD%':>7s}{'Exp%':>7s}{'AnniPos':>9s}")
|
||||
print(" " + "-" * 118)
|
||||
for label, (fn, params) in CANDIDATES.items():
|
||||
for asset in ["BTC", "ETH"]:
|
||||
for tf in ["1h", "4h"]:
|
||||
df = get_df(asset, tf)
|
||||
ents = fn(df, **params)
|
||||
full = simulate(ents, df)
|
||||
split = int(len(df) * (1 - OOS_FRAC))
|
||||
oos = simulate([e for e in ents if e["i"] >= split], df)
|
||||
yrs = full["yearly"]
|
||||
pos_yrs = sum(1 for v in yrs.values() if v > 0)
|
||||
tr_yr = full["trades"] / max(len(yrs), 1)
|
||||
robust = oos["ret"] > 0 and full["ret"] > 0 and pos_yrs >= max(len(yrs) - 1, 1)
|
||||
flag = " <<<" if robust else ""
|
||||
print(f" {label:<22s}{asset:>5s}{tf:>5s}{full['trades']:>6d}{tr_yr:>7.0f}{full['win']:>7.1f}"
|
||||
f"{full['ret']:>+10.1f}{oos['ret']:>+10.1f}{full['dd']:>7.1f}{full['exposure']:>7.1f}"
|
||||
f"{f'{pos_yrs}/{len(yrs)}':>9s}{flag}")
|
||||
print(" " + "-" * 118)
|
||||
print(" <<< = positivo full+OOS e robusto (quasi tutti gli anni positivi).")
|
||||
|
||||
|
||||
def deep_dive():
|
||||
"""Robustezza dei 3 candidati promossi: fee sweep + griglia parametri OOS."""
|
||||
split_of = lambda df: int(len(df) * (1 - OOS_FRAC))
|
||||
fees = [0.0, 0.0005, 0.001, 0.002]
|
||||
|
||||
print("\n" + "#" * 122)
|
||||
print(" APPROFONDIMENTO MR02 / MR03 / MR05 — robustezza fee + griglia (deve restare positivo)")
|
||||
print("#" * 122)
|
||||
|
||||
# --- MR02 Donchian Fade ---
|
||||
print(f"\n [MR02 donchian_fade] SENSIBILITA' FEE — Ret% FULL/OOS (n=20)")
|
||||
print(f" {'Asset/TF':<10s}" + "".join(f"{f'{f*100:.2f}%RT':>22s}" for f in fees))
|
||||
print(f" {'':<10s}" + "".join(f"{'full':>11s}{'oos':>11s}" for _ in fees))
|
||||
for a, tf in [("BTC", "1h"), ("ETH", "1h"), ("BTC", "4h"), ("ETH", "4h")]:
|
||||
df = get_df(a, tf); sp = split_of(df)
|
||||
ents = donchian_fade(df, n=20, sl_atr=2.0, max_bars=24)
|
||||
oents = [e for e in ents if e["i"] >= sp]
|
||||
cells = "".join(f"{simulate(ents, df, fee_rt=f)['ret']:>+11.0f}{simulate(oents, df, fee_rt=f)['ret']:>+11.0f}" for f in fees)
|
||||
print(f" {a+' '+tf:<10s}{cells}")
|
||||
print(f"\n [MR02] GRIGLIA n x sl_atr — Ret%OOS(DD%) | fee {FEE_RT*100:.2f}% RT")
|
||||
for a in ["BTC", "ETH"]:
|
||||
df = get_df(a, "1h"); sp = split_of(df)
|
||||
print(f"\n {a} 1h " + "".join(f"{f'sl={s}':>16s}" for s in [1.5, 2.0, 3.0]))
|
||||
for n in [10, 20, 30, 50]:
|
||||
cells = ""
|
||||
for s in [1.5, 2.0, 3.0]:
|
||||
r = simulate([e for e in donchian_fade(df, n=n, sl_atr=s, max_bars=24) if e["i"] >= sp], df)
|
||||
cell = "%+.0f(%.0f)" % (r["ret"], r["dd"])
|
||||
cells += f"{cell:>16s}"
|
||||
print(f" n={n:<4d}{cells}")
|
||||
|
||||
# --- MR03 Keltner Fade ---
|
||||
print(f"\n [MR03 keltner_fade] GRIGLIA n x k — Ret%OOS(DD%) | fee {FEE_RT*100:.2f}% RT")
|
||||
for a in ["BTC", "ETH"]:
|
||||
df = get_df(a, "1h"); sp = split_of(df)
|
||||
print(f"\n {a} 1h " + "".join(f"{f'k={k}':>16s}" for k in [1.5, 2.0, 2.5]))
|
||||
for n in [14, 20, 30, 50]:
|
||||
cells = ""
|
||||
for k in [1.5, 2.0, 2.5]:
|
||||
r = simulate([e for e in keltner_fade(df, n=n, k=k, sl_atr=2.0, max_bars=24) if e["i"] >= sp], df)
|
||||
cell = "%+.0f(%.0f)" % (r["ret"], r["dd"])
|
||||
cells += f"{cell:>16s}"
|
||||
print(f" n={n:<4d}{cells}")
|
||||
|
||||
# --- MR05 Bollinger Fade + ADX ---
|
||||
print(f"\n [MR05 boll_fade_adx] GRIGLIA n x adx_max — Ret%OOS(DD%) | fee {FEE_RT*100:.2f}% RT")
|
||||
for a in ["BTC", "ETH"]:
|
||||
df = get_df(a, "1h"); sp = split_of(df)
|
||||
print(f"\n {a} 1h " + "".join(f"{f'adx<{x}':>16s}" for x in [20, 25, 30]))
|
||||
for n in [20, 30, 50]:
|
||||
cells = ""
|
||||
for x in [20, 25, 30]:
|
||||
r = simulate([e for e in boll_fade_adx(df, n=n, k=2.5, sl_atr=2.0, max_bars=24, adx_max=x) if e["i"] >= sp], df)
|
||||
cell = "%+.0f(%.0f)" % (r["ret"], r["dd"])
|
||||
cells += f"{cell:>16s}"
|
||||
print(f" n={n:<4d}{cells}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
table()
|
||||
deep_dive()
|
||||
@@ -57,16 +57,21 @@ class BollingerFade(Strategy):
|
||||
k = params.get("k", 2.5)
|
||||
sl_atr = params.get("sl_atr", 2.0)
|
||||
max_bars = params.get("max_bars", 24)
|
||||
trend_max = params.get("trend_max") # None = filtro disattivo
|
||||
ema_long = params.get("ema_long", 200)
|
||||
|
||||
ma = pd.Series(c).rolling(bb_w).mean().values
|
||||
sd = pd.Series(c).rolling(bb_w).std().values
|
||||
a = _atr(df, 14)
|
||||
up, lo = ma + k * sd, ma - k * sd
|
||||
el = pd.Series(c).ewm(span=ema_long, adjust=False).mean().values if trend_max is not None else None
|
||||
|
||||
signals: list[Signal] = []
|
||||
for i in range(bb_w + 14, n_len):
|
||||
if np.isnan(up[i]) or np.isnan(a[i]):
|
||||
continue
|
||||
if el is not None and (a[i] == 0 or np.isnan(el[i]) or abs(c[i] - el[i]) / a[i] > trend_max):
|
||||
continue
|
||||
if c[i] < lo[i] and c[i - 1] >= lo[i - 1]:
|
||||
d, sl = 1, c[i] - sl_atr * a[i]
|
||||
elif c[i] > up[i] and c[i - 1] <= up[i - 1]:
|
||||
|
||||
@@ -0,0 +1,82 @@
|
||||
"""MR02 — Donchian Fade (mean-reversion sugli estremi del canale).
|
||||
|
||||
L'opposto esatto del trend-following Donchian (che PERDE netto: vedi
|
||||
scripts/analysis/strategy_research.py). Coerente con la lezione squeeze:
|
||||
i breakout RIENTRANO, quindi si fada la rottura del canale verso il centro.
|
||||
|
||||
Logica:
|
||||
1. Canale Donchian: massimo/minimo delle ultime n barre (escludendo la corrente)
|
||||
2. ENTRY: close rompe SOPRA il massimo del canale -> SHORT (fade);
|
||||
close rompe SOTTO il minimo -> LONG. Ingresso a close[i] (eseguibile).
|
||||
3. EXIT: take-profit al centro del canale (il rientro atteso),
|
||||
stop-loss a sl_atr*ATR oltre l'estremo, time-limit max_bars.
|
||||
|
||||
Validazione (netto, fee 0.10% RT reale Deribit, leva 3x, OOS = ultimo 30%):
|
||||
BTC 1h n=20: +879% FULL / +171% OOS, DD 30%, 8/9 anni positivi
|
||||
ETH 1h n=20: enorme FULL / +8452% OOS, DD 42%
|
||||
Robusto su TUTTA la griglia n in {10,20,30,50} x sl_atr in {1.5,2.0,3.0}
|
||||
(BTC+ETH 1h sempre positivo OOS) e su tutte le fee 0.00-0.20% RT.
|
||||
Ricerca completa: scripts/analysis/strategy_research_v2.py.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
import sys
|
||||
sys.path.insert(0, ".")
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
|
||||
from src.strategies.base import Signal
|
||||
from src.strategies.fade_base import FadeStrategy, atr, trend_distance
|
||||
|
||||
|
||||
class DonchianFade(FadeStrategy):
|
||||
name = "MR02_donchian_fade"
|
||||
description = "Mean-reversion: fada la rottura del canale Donchian, TP al centro"
|
||||
default_assets = ["BTC", "ETH"]
|
||||
default_timeframes = ["1h"]
|
||||
|
||||
def generate_signals(self, df: pd.DataFrame, ts: pd.DatetimeIndex,
|
||||
**params) -> list[Signal]:
|
||||
n = params.get("n", 20)
|
||||
sl_atr = params.get("sl_atr", 2.0)
|
||||
max_bars = params.get("max_bars", 24)
|
||||
trend_max = params.get("trend_max") # None = filtro disattivo
|
||||
ema_long = params.get("ema_long", 200)
|
||||
|
||||
h, l, c = df["high"].values, df["low"].values, df["close"].values
|
||||
hh = pd.Series(h).rolling(n).max().shift(1).values
|
||||
ll = pd.Series(l).rolling(n).min().shift(1).values
|
||||
a = atr(df, 14)
|
||||
td = trend_distance(df, ema_long) if trend_max is not None else None
|
||||
|
||||
signals: list[Signal] = []
|
||||
for i in range(n + 14, len(c)):
|
||||
if np.isnan(hh[i]) or np.isnan(a[i]):
|
||||
continue
|
||||
if td is not None and (np.isnan(td[i]) or td[i] > trend_max):
|
||||
continue
|
||||
mid = (hh[i] + ll[i]) / 2.0
|
||||
if c[i] > hh[i] and c[i - 1] <= hh[i - 1]: # rottura rialzista -> fade short
|
||||
d, sl = -1, c[i] + sl_atr * a[i]
|
||||
elif c[i] < ll[i] and c[i - 1] >= ll[i - 1]: # rottura ribassista -> fade long
|
||||
d, sl = 1, c[i] - sl_atr * a[i]
|
||||
else:
|
||||
continue
|
||||
signals.append(Signal(
|
||||
idx=i, direction=d, entry_price=c[i],
|
||||
metadata={"tp": float(mid), "sl": float(sl), "max_bars": max_bars},
|
||||
))
|
||||
return signals
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
strat = DonchianFade()
|
||||
print("=" * 110)
|
||||
print(f" MR02 DONCHIAN FADE — netto fee {strat.fee_rt*100:.2f}% RT, leva {strat.leverage:.0f}x")
|
||||
print("=" * 110)
|
||||
for asset in ["BTC", "ETH"]:
|
||||
r = strat.backtest(asset, "1h", n=20, sl_atr=2.0, max_bars=24)
|
||||
if r:
|
||||
r.strategy_name = f"MR02 {asset} 1h n20"
|
||||
r.print_summary()
|
||||
r.print_yearly()
|
||||
@@ -0,0 +1,82 @@
|
||||
"""MR03 — Keltner Fade (mean-reversion sul canale ATR).
|
||||
|
||||
Stessa tesi di MR01 (i breakout rientrano) ma con banda costruita su ATR
|
||||
attorno a una EMA, invece che su deviazione standard attorno a una SMA.
|
||||
Reagisce diversamente a gap e code: edge indipendente, non ridondante con MR01.
|
||||
|
||||
Logica:
|
||||
1. Canale di Keltner: EMA(n) +/- k*ATR(n)
|
||||
2. ENTRY: close esce sotto la banda inferiore -> LONG (o sopra la superiore -> SHORT)
|
||||
Ingresso a close[i] (eseguibile dal vivo, nessun look-ahead).
|
||||
3. EXIT: take-profit alla EMA centrale (il rientro atteso),
|
||||
stop-loss a sl_atr*ATR oltre l'estremo, time-limit max_bars.
|
||||
|
||||
Validazione (netto, fee 0.10% RT reale Deribit, leva 3x, OOS = ultimo 30%):
|
||||
BTC 1h n=30 k=2.0: +112% OOS, DD 20%
|
||||
ETH 1h n=50 k=1.5: +1426% OOS, DD 20%
|
||||
Robusto su TUTTA la griglia n in {14,20,30,50} x k in {1.5,2.0,2.5}
|
||||
(BTC+ETH 1h sempre positivo OOS).
|
||||
Ricerca completa: scripts/analysis/strategy_research_v2.py.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
import sys
|
||||
sys.path.insert(0, ".")
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
|
||||
from src.strategies.base import Signal
|
||||
from src.strategies.fade_base import FadeStrategy, atr, trend_distance
|
||||
|
||||
|
||||
class KeltnerFade(FadeStrategy):
|
||||
name = "MR03_keltner_fade"
|
||||
description = "Mean-reversion: fada il canale di Keltner (ATR), TP alla EMA"
|
||||
default_assets = ["BTC", "ETH"]
|
||||
default_timeframes = ["1h"]
|
||||
|
||||
def generate_signals(self, df: pd.DataFrame, ts: pd.DatetimeIndex,
|
||||
**params) -> list[Signal]:
|
||||
n = params.get("n", 30)
|
||||
k = params.get("k", 2.0)
|
||||
sl_atr = params.get("sl_atr", 2.0)
|
||||
max_bars = params.get("max_bars", 24)
|
||||
trend_max = params.get("trend_max") # None = filtro disattivo
|
||||
ema_long = params.get("ema_long", 200)
|
||||
|
||||
c = df["close"].values
|
||||
e = pd.Series(c).ewm(span=n, adjust=False).mean().values
|
||||
a = atr(df, n)
|
||||
up, lo = e + k * a, e - k * a
|
||||
td = trend_distance(df, ema_long) if trend_max is not None else None
|
||||
|
||||
signals: list[Signal] = []
|
||||
for i in range(n + 1, len(c)):
|
||||
if np.isnan(up[i]) or np.isnan(a[i]):
|
||||
continue
|
||||
if td is not None and (np.isnan(td[i]) or td[i] > trend_max):
|
||||
continue
|
||||
if c[i] < lo[i] and c[i - 1] >= lo[i - 1]:
|
||||
d, sl = 1, c[i] - sl_atr * a[i]
|
||||
elif c[i] > up[i] and c[i - 1] <= up[i - 1]:
|
||||
d, sl = -1, c[i] + sl_atr * a[i]
|
||||
else:
|
||||
continue
|
||||
signals.append(Signal(
|
||||
idx=i, direction=d, entry_price=c[i],
|
||||
metadata={"tp": float(e[i]), "sl": float(sl), "max_bars": max_bars},
|
||||
))
|
||||
return signals
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
strat = KeltnerFade()
|
||||
print("=" * 110)
|
||||
print(f" MR03 KELTNER FADE — netto fee {strat.fee_rt*100:.2f}% RT, leva {strat.leverage:.0f}x")
|
||||
print("=" * 110)
|
||||
for asset, n, k in [("BTC", 30, 2.0), ("ETH", 50, 1.5)]:
|
||||
r = strat.backtest(asset, "1h", n=n, k=k, sl_atr=2.0, max_bars=24)
|
||||
if r:
|
||||
r.strategy_name = f"MR03 {asset} 1h n{n} k{k}"
|
||||
r.print_summary()
|
||||
r.print_yearly()
|
||||
@@ -0,0 +1,88 @@
|
||||
"""MR07 — Return Reversal (fade del movimento di barra estremo).
|
||||
|
||||
Meccanismo distinto da MR01/MR02/MR03: non guarda i LIVELLI di prezzo (bande,
|
||||
canali) ma la VOLATILITA' dei rendimenti. Quando una singola barra si muove di
|
||||
piu' di k deviazioni standard rolling dei rendimenti, e' un'over-reaction che
|
||||
tende a rientrare: si fada nella direzione opposta. Coerente con la lezione
|
||||
mean-reversion.
|
||||
|
||||
Logica:
|
||||
1. ret[i] = rendimento dell'ultima barra; sigma = std rolling(n) dei rendimenti
|
||||
2. z = ret[i]/sigma. Se z <= -k (crollo) -> LONG; se z >= +k (spike) -> SHORT.
|
||||
Ingresso a close[i] (eseguibile dal vivo, nessun look-ahead).
|
||||
3. EXIT: take-profit a tp_atr*ATR a favore, stop-loss a sl_atr*ATR contro,
|
||||
time-limit max_bars.
|
||||
|
||||
Validazione (netto, fee 0.10% RT reale Deribit, leva 3x, OOS = ultimo 30%):
|
||||
config robusta k=3.5 tp=2ATR sl=1.5ATR n=50:
|
||||
BTC 1h: +447% FULL / +105% OOS, DD 25%
|
||||
ETH 1h: +335% FULL / +195% OOS, DD 46%
|
||||
L'intero blocco tp_atr=2.0 (k in {2.5,3.0,3.5} x sl in {1.5,2.0,2.5}) e'
|
||||
positivo full+OOS su entrambi gli asset 1h.
|
||||
Ricerca completa: scripts/analysis/strategy_research_v2.py.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
import sys
|
||||
sys.path.insert(0, ".")
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
|
||||
from src.strategies.base import Signal
|
||||
from src.strategies.fade_base import FadeStrategy, atr, trend_distance
|
||||
|
||||
|
||||
class ReturnReversal(FadeStrategy):
|
||||
name = "MR07_return_reversal"
|
||||
description = "Mean-reversion: fada il movimento di barra estremo (z dei rendimenti)"
|
||||
default_assets = ["BTC", "ETH"]
|
||||
default_timeframes = ["1h"]
|
||||
|
||||
def generate_signals(self, df: pd.DataFrame, ts: pd.DatetimeIndex,
|
||||
**params) -> list[Signal]:
|
||||
n = params.get("n", 50)
|
||||
k = params.get("k", 3.5)
|
||||
tp_atr = params.get("tp_atr", 2.0)
|
||||
sl_atr = params.get("sl_atr", 1.5)
|
||||
max_bars = params.get("max_bars", 24)
|
||||
trend_max = params.get("trend_max") # None = filtro disattivo
|
||||
ema_long = params.get("ema_long", 200)
|
||||
|
||||
c = df["close"].values
|
||||
ret = np.zeros_like(c)
|
||||
ret[1:] = np.diff(c) / c[:-1]
|
||||
sig = pd.Series(ret).rolling(n).std().values
|
||||
a = atr(df, 14)
|
||||
td = trend_distance(df, ema_long) if trend_max is not None else None
|
||||
|
||||
signals: list[Signal] = []
|
||||
for i in range(n + 14, len(c)):
|
||||
if np.isnan(sig[i]) or sig[i] == 0 or np.isnan(a[i]):
|
||||
continue
|
||||
if td is not None and (np.isnan(td[i]) or td[i] > trend_max):
|
||||
continue
|
||||
z = ret[i] / sig[i]
|
||||
if z <= -k: # crollo di barra -> fade long
|
||||
d, tp, sl = 1, c[i] + tp_atr * a[i], c[i] - sl_atr * a[i]
|
||||
elif z >= k: # spike di barra -> fade short
|
||||
d, tp, sl = -1, c[i] - tp_atr * a[i], c[i] + sl_atr * a[i]
|
||||
else:
|
||||
continue
|
||||
signals.append(Signal(
|
||||
idx=i, direction=d, entry_price=c[i],
|
||||
metadata={"tp": float(tp), "sl": float(sl), "max_bars": max_bars},
|
||||
))
|
||||
return signals
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
strat = ReturnReversal()
|
||||
print("=" * 110)
|
||||
print(f" MR07 RETURN REVERSAL — netto fee {strat.fee_rt*100:.2f}% RT, leva {strat.leverage:.0f}x")
|
||||
print("=" * 110)
|
||||
for asset in ["BTC", "ETH"]:
|
||||
r = strat.backtest(asset, "1h", n=50, k=3.5, tp_atr=2.0, sl_atr=1.5, max_bars=24)
|
||||
if r:
|
||||
r.strategy_name = f"MR07 {asset} 1h k3.5"
|
||||
r.print_summary()
|
||||
r.print_yearly()
|
||||
Reference in New Issue
Block a user