test(live): demo numerica exit intrabar fade -- worker replay ~= backtest

MR01 BTC 1h, 4000 barre, no filtro trend: backtest build_trades +3.5%/73 trade vs
worker replay intrabar +4.5%/78 trade -> gap +1.0pt (allineato). Conferma che il fix
exit intrabar chiude il gap live-vs-backtest delle fade (residuo = bar-timing).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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()