feat(research): FR01 Hurst-Calm Fade + analisi per-anno/mercato (ricerca 100 agenti)
Esito ricerca frattali x regime ARGO (171 agenti): - FR01_hurst_calm_fade.py: vincitore = fade gateato da hurst<0.55 (anti-persistente) + dvol_pct<0.4 (DVOL bassa). OOS Sharpe 3.73 BTC, 6/6 anni positivi su BTC+ETH, corr bassa coi fade esistenti (MR01 +0.17/MR02 +0.08/MR07 -0.03) -> diversificatore non ridondante. - fractal_argo_peryear.py: analisi per-anno/regime-mercato dei top candidati. - diario 2026-06-02: verdetto completo. Finding chiave: prior ARGO 'VRP>0=range=fade' SMENTITO, l'edge robusto e' su VRP<0 + DVOL bassa. Diversificatori, NON spodestano PORT06 (OOS Sharpe 8.19). Branch di ricerca. Deploy bloccato da: verifica corr sul MASTER intero + wiring DVOL live nel runner. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Adriano Dal Pastro
@@ -0,0 +1,86 @@
|
||||
import sys; sys.path.insert(0,".")
|
||||
import numpy as np, pandas as pd
|
||||
from scripts.analysis.regime_lab import load_features
|
||||
from scripts.analysis.explore_lab import atr
|
||||
|
||||
FEE=0.001; LEV=3
|
||||
|
||||
def build(df, gate, k=2.5, sl_atr=2.0, mb=24, bb=50):
|
||||
c=df['close'].values; a=atr(df,14)
|
||||
ma=pd.Series(c).rolling(bb).mean().values; sd=pd.Series(c).rolling(bb).std().values
|
||||
ent=[]
|
||||
for i in range(bb+14,len(c)-1):
|
||||
if np.isnan(sd[i]) or sd[i]==0 or np.isnan(a[i]): continue
|
||||
if not gate(df,i): continue
|
||||
if c[i]<ma[i]-k*sd[i]: d,sl=1,c[i]-sl_atr*a[i]
|
||||
elif c[i]>ma[i]+k*sd[i]: d,sl=-1,c[i]+sl_atr*a[i]
|
||||
else: continue
|
||||
ent.append({'i':i,'d':d,'tp':ma[i],'sl':sl,'mb':mb})
|
||||
return ent
|
||||
|
||||
def per_year(df, ent):
|
||||
"""replay intrabar fedele (sl-first, tp, poi max_bars@close) -> per anno {n,ret%,win%}."""
|
||||
h=df['high'].values; l=df['low'].values; c=df['close'].values
|
||||
ts=pd.to_datetime(df['timestamp'],unit='ms',utc=True)
|
||||
Y={}
|
||||
last=-1
|
||||
for e in ent:
|
||||
i=e['i']
|
||||
if i<=last: continue
|
||||
d=e['d']; tp=e['tp']; sl=e['sl']; j=min(i+e['mb'],len(c)-1)
|
||||
exit_p=c[j]
|
||||
for t in range(i+1,j+1):
|
||||
if d==1:
|
||||
if l[t]<=sl: exit_p=sl; j=t; break
|
||||
if h[t]>=tp: exit_p=tp; j=t; break
|
||||
else:
|
||||
if h[t]>=sl: exit_p=sl; j=t; break
|
||||
if l[t]<=tp: exit_p=tp; j=t; break
|
||||
ret=(exit_p-c[i])/c[i]*d*LEV - FEE*LEV
|
||||
last=j; yr=ts.iloc[i].year
|
||||
if yr not in Y: Y[yr]=[0,0.0,0]
|
||||
Y[yr][0]+=1; Y[yr][1]+=ret*100; Y[yr][2]+= (ret>0)
|
||||
return Y
|
||||
|
||||
# gate functions
|
||||
def g_hurst_calm(df,i): return df['hurst'].iloc[i]<0.55 and not np.isnan(df['dvol_pct'].iloc[i]) and df['dvol_pct'].iloc[i]<0.40
|
||||
def g_vrp_neg(df,i): return not np.isnan(df['vrp'].iloc[i]) and df['vrp'].iloc[i]<0
|
||||
def g_hig_vrp(df,i):
|
||||
hi=df['higuchi'].iloc[i]; return (not np.isnan(hi)) and hi>1.5 and (not np.isnan(df['vrp'].iloc[i])) and df['vrp'].iloc[i]<0
|
||||
def g_none(df,i): return True
|
||||
|
||||
STRATS=[("HurstCalmFade (hurst<.55 & DVOL<p40)",g_hurst_calm),
|
||||
("VRP<0 Fade (core driver)",g_vrp_neg),
|
||||
("HigVRP Fade (Higuchi>1.5 & VRP<0)",g_hig_vrp),
|
||||
("Fade NUDA (no gate, baseline)",g_none)]
|
||||
|
||||
# regime mercato BTC per anno (da BTC close annuale)
|
||||
btc=load_features("BTC","1d")
|
||||
bts=pd.to_datetime(btc['timestamp'],unit='ms',utc=True); bc=btc['close'].values
|
||||
mkt={}
|
||||
for yr in range(2021,2027):
|
||||
m=(bts.dt.year==yr).values
|
||||
if m.sum()>5:
|
||||
r=(bc[m][-1]/bc[m][0]-1)*100
|
||||
mkt[yr]=("BULL" if r>40 else "BEAR" if r<-30 else "RANGE", r)
|
||||
|
||||
for asset in ("BTC","ETH"):
|
||||
df=load_features(asset,"1h")
|
||||
print(f"\n{'='*78}\n {asset} 1h — performance per anno (somma ret% per-trade, netto leva3x+fee0.10%)\n{'='*78}")
|
||||
print(f" {'Strategia':<40} " + " ".join(f"{y}" for y in range(2021,2027)))
|
||||
for name,g in STRATS:
|
||||
ent=build(df,g); Y=per_year(df,ent)
|
||||
cells=[]
|
||||
for y in range(2021,2027):
|
||||
if y in Y and Y[y][0]>0:
|
||||
cells.append(f"{Y[y][1]:+5.0f}")
|
||||
else: cells.append(" . ")
|
||||
print(f" {name:<40} " + " ".join(cells))
|
||||
# riga trades/anno per la strategia principale
|
||||
ent=build(df,g_hurst_calm); Y=per_year(df,ent)
|
||||
tr=" ".join(f"{Y.get(y,[0])[0]:>5}" for y in range(2021,2027))
|
||||
print(f" {'(HurstCalmFade trades/anno)':<40} {tr}")
|
||||
|
||||
print(f"\n REGIME MERCATO BTC per anno (ret% annuale prezzo):")
|
||||
for y in range(2021,2027):
|
||||
if y in mkt: print(f" {y}: {mkt[y][0]:6} ({mkt[y][1]:+.0f}%)")
|
||||
@@ -0,0 +1,123 @@
|
||||
"""FR01 — Hurst-Calm Fade (FRATTALE x REGIME). Esito della ricerca a 100 agenti (2026-06-02).
|
||||
|
||||
Fade della banda di Bollinger (k=2.5 su SMA50, TP=media, SL=2*ATR, max_bars=24) ATTIVATO
|
||||
SOLO quando coincidono due condizioni di regime:
|
||||
- FRATTALE: rolling Hurst < 0.55 (regime anti-persistente -> la mean-reversion ha senso
|
||||
fratalmente; con Hurst>0.55 il fade peggiora, il momentum perde comunque).
|
||||
- VOLATILITA: dvol_pct < 0.40 (DVOL nel terzile basso del suo storico -> regime calmo/range).
|
||||
|
||||
Doppio gate frattale x regime: l'INTERAZIONE e' l'ingrediente attivo, non il fade di per se'
|
||||
(ablation: senza gate Sharpe ~0.8 e muore a fee 0.2% RT; col doppio gate OOS Sharpe ~3.7).
|
||||
|
||||
VALIDAZIONE (netto fee 0.10% RT, leva 3x, OOS ultimo 30%, ricerca fractal_argo_workflow):
|
||||
BTC 1h: 198 trade, FULL +100% / OOS +54% / Sharpe OOS 3.73 / DD OOS 5.1% / 6/6 anni positivi,
|
||||
regge fee 0.2% RT. Confermato avversarialmente (no look-ahead, split alternativo).
|
||||
Generalizza a ETH 1h (Sharpe ~2.6, secondario). 4h/1d = rumore (pochi trade).
|
||||
Correlazione coi fade esistenti BASSA: MR01 +0.17, MR02 +0.08, MR07 -0.03 -> DIVERSIFICATORE
|
||||
quasi-ortogonale (profilo SH01/pairs), NON ridondante. Esposizione ~1-9% -> low-frequency.
|
||||
|
||||
RUOLO: diversificatore a basso DD per il MASTER/PORT06, NON motore standalone (non batte il
|
||||
portafoglio da solo). Coerente coi priori: i frattali da soli sono rumore; il valore e' nel
|
||||
gating del regime. NB il prior ARGO "VRP>0=range=fade" e' SMENTITO: l'edge robusto e' su VRP<0
|
||||
e su DVOL bassa (questo gate dvol_pct<0.4), non su vol alta.
|
||||
|
||||
DIPENDENZA REGIME (caveat deploy): il gate usa DVOL/dvol_pct. Per il BACKTEST le feature
|
||||
arrivano da regime_lab (cache da Deribit mainnet). Per il LIVE serve un feed DVOL in produzione
|
||||
(regime_fetcher + allineamento causale nel runner) -> wiring NON ancora fatto. Finche' manca,
|
||||
FR01 e' validata-in-ricerca ma non deployabile live.
|
||||
|
||||
Run backtest: uv run python scripts/strategies/FR01_hurst_calm_fade.py
|
||||
"""
|
||||
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.strategies.base import Strategy, Signal # noqa: E402
|
||||
from src.fractal.indicators import rolling_hurst # noqa: E402
|
||||
|
||||
|
||||
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 HurstCalmFade(Strategy):
|
||||
name = "FR01_hurst_calm_fade"
|
||||
description = "Fade Bollinger gateato da Hurst<0.55 (anti-persistente) + DVOL bassa (calm)"
|
||||
default_assets = ["BTC", "ETH"]
|
||||
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]:
|
||||
bb_w = params.get("bb_window", 50)
|
||||
k = params.get("k", 2.5)
|
||||
sl_atr = params.get("sl_atr", 2.0)
|
||||
max_bars = params.get("max_bars", 24)
|
||||
hurst_thr = params.get("hurst_thr", 0.55)
|
||||
hurst_win = params.get("hurst_win", 100)
|
||||
dvol_pct_thr = params.get("dvol_pct_thr", 0.40)
|
||||
|
||||
c = df["close"].values.astype(float)
|
||||
n = len(c)
|
||||
ma = pd.Series(c).rolling(bb_w).mean().values
|
||||
sd = pd.Series(c).rolling(bb_w).std().values
|
||||
a = _atr(df, 14)
|
||||
hurst = rolling_hurst(c, window=hurst_win) # causale (returns[i-win:i])
|
||||
# dvol_pct: dalla colonna se presente (regime_lab.load_features), altrimenti gate OFF
|
||||
dvol_pct = df["dvol_pct"].values if "dvol_pct" in df.columns else np.full(n, np.nan)
|
||||
|
||||
signals: list[Signal] = []
|
||||
for i in range(bb_w + 14, n):
|
||||
if np.isnan(sd[i]) or np.isnan(a[i]) or sd[i] == 0:
|
||||
continue
|
||||
# GATE FRATTALE x REGIME (tutto noto a i)
|
||||
if hurst[i] >= hurst_thr:
|
||||
continue
|
||||
if "dvol_pct" in df.columns:
|
||||
if np.isnan(dvol_pct[i]) or dvol_pct[i] >= dvol_pct_thr:
|
||||
continue
|
||||
up, lo = ma[i] + k * sd[i], ma[i] - k * sd[i]
|
||||
if c[i] < lo:
|
||||
d, sl = 1, c[i] - sl_atr * a[i]
|
||||
elif c[i] > up:
|
||||
d, sl = -1, c[i] + sl_atr * a[i]
|
||||
else:
|
||||
continue
|
||||
signals.append(Signal(
|
||||
idx=i, direction=d, entry_price=float(c[i]),
|
||||
metadata={"tp": float(ma[i]), "sl": float(sl), "max_bars": int(max_bars)},
|
||||
))
|
||||
return signals
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# backtest via l'harness onesto + feature di regime_lab (DVOL reale)
|
||||
from scripts.analysis.regime_lab import load_features, report
|
||||
from scripts.analysis.explore_lab import robust
|
||||
|
||||
strat = HurstCalmFade()
|
||||
print(f"{'=' * 100}")
|
||||
print(f" FR01 HURST-CALM FADE — netto fee {strat.fee_rt*100:.2f}% RT, leva {strat.leverage:.0f}x")
|
||||
print(f" gate: hurst<0.55 (anti-persistente) + dvol_pct<0.40 (DVOL bassa)")
|
||||
print(f"{'=' * 100}")
|
||||
for asset in ("BTC", "ETH"):
|
||||
df = load_features(asset, "1h")
|
||||
ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
|
||||
sigs = strat.generate_signals(df, ts)
|
||||
ent = [{"i": s.idx, "d": s.direction, "tp": s.metadata["tp"],
|
||||
"sl": s.metadata["sl"], "max_bars": s.metadata["max_bars"]} for s in sigs]
|
||||
res = report(f"FR01_{asset}_1h", ent, df)
|
||||
print(f" -> {asset}: robust={robust(res)} OOS Sharpe={res['oos']['sharpe']:.2f} "
|
||||
f"OOS ret={res['oos']['ret']:+.0f}% DD={res['full']['dd']:.0f}%")
|
||||
Reference in New Issue
Block a user