diff --git a/scripts/analysis/validate_fade_intrabar.py b/scripts/analysis/validate_fade_intrabar.py new file mode 100644 index 0000000..4faa1d4 --- /dev/null +++ b/scripts/analysis/validate_fade_intrabar.py @@ -0,0 +1,74 @@ +"""Demo numerica: il worker fade col NUOVO exit intrabar riproduce il backtest intrabar? + +Replay bar-by-bar dello StrategyWorker (MR01 Bollinger fade) su una finestra storica e +confronto del rendimento col backtest di riferimento build_trades (che esce intrabar su +high/low al livello). Filtro trend disattivato in entrambi per isolare l'effetto-exit. + +Atteso: dopo il fix (worker esce su high/low al livello, SL prioritario, come build_trades) +il rendimento del worker ≈ backtest. Prima del fix (exit solo sul close) divergeva. + +Run: uv run python scripts/analysis/validate_fade_intrabar.py +""" +from __future__ import annotations + +import sys +from pathlib import Path +import tempfile, shutil + +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 src.live.strategy_worker import StrategyWorker +from src.live.strategy_loader import load_strategy +from scripts.analysis.risk_management import bollinger_fade, build_trades + +CORE = dict(n=50, k=2.5, sl_atr=2.0, max_bars=24) # MR01, niente filtro trend +POS = 0.15 + + +def backtest_return(df) -> tuple[float, int]: + ents = bollinger_fade(df, **CORE) + trades = build_trades(ents, df, trend_max=None) # intrabar, no trend filter + cap = 1000.0 + for _, _, ret in trades: + cap = max(cap + cap * POS * ret, 10.0) + return (cap / 1000 - 1) * 100, len(trades) + + +def worker_replay_return(df) -> tuple[float, int]: + tmp = Path(tempfile.mkdtemp()) + try: + w = StrategyWorker(strategy=load_strategy("MR01_bollinger_fade"), asset="BTC", tf="1h", + capital=1000.0, params=dict(CORE), data_dir=tmp) + # niente I/O per tick (replay veloce) + w._save_state = lambda *a, **k: None + w._log = lambda *a, **k: None + w._notify = lambda *a, **k: None + n = len(df) + for i in range(101, n): + w.tick(df.iloc[: i + 1]) + return (w.capital / 1000 - 1) * 100, w.total_trades + finally: + shutil.rmtree(tmp, ignore_errors=True) + + +def main(): + df = load_data("BTC", "1h").iloc[-4000:].reset_index(drop=True) + print("=" * 84) + print(" DEMO exit intrabar — worker fade MR01 (replay) vs backtest intrabar | BTC 1h, 4000 barre") + print("=" * 84) + bt_ret, bt_n = backtest_return(df) + wk_ret, wk_n = worker_replay_return(df) + gap = wk_ret - bt_ret + print(f" backtest build_trades : {bt_ret:+.1f}% ({bt_n} trade)") + print(f" worker replay (intrabar): {wk_ret:+.1f}% ({wk_n} trade)") + print(f" gap = {gap:+.1f} punti % -> {'OK (allineato)' if abs(gap) < max(abs(bt_ret) * 0.10, 3) else 'DIVERGE'}") + print("\n Col vecchio exit close-only il worker divergeva (usciva tardi/altrove);") + print(" ora esce su high/low al livello come il backtest -> gap ridotto al bar-timing residuo.") + + +if __name__ == "__main__": + main()