Adriano Dal Pastro
14522262e6
Reset del progetto su fondamenta verificate dopo la scoperta che l'intera libreria "validata OOS" era artefatto di feed contaminato (print fantasma del feed Cerbero TESTNET + storico Binance/USDT). - Storico ricostruito da Deribit MAINNET (ccxt pubblico, tokenless) e CERTIFICATO (certify_feed.py): BTC/ETH puliti su TUTTA la storia (mediana 2-6 bps vs Coinbase USD), integrita' OHLC + coerenza resample (maxΔ 0.00) + cross-venue OK. Alt esclusi (illiquidi/divergenti: LTC/DOGE 50-82% barre flat; XRP/BNB non certificabili). - Verdetto sul feed pulito: FADE / PAIRS / XS01 / TSM01 morti (ogni portafoglio Sharpe -2.3..-3.0, DD ~40%); solo SH01 e frammenti HONEST con segnale residuo, da ri-validare in isolamento. - Cleanup "restart pulito": strategie, stack live (src/live, src/portfolio, runner/executor, yml, docker), ~100 script ricerca/gate, waste/games/ portfolios, dati non certificati + cache e 60+ diari -> archiviati in Old/ (preservati, non cancellati). Diario consolidato in un unico documento. - Skeleton ricerca tenuto: Strategy ABC + indicatori + src/fractal + src/backtest/engine + load_data; tool dati certificati (rebuild_history, certify_feed, audit_feed, multi_source_check). - Universo dati ATTIVO: solo BTC/ETH (5m/15m/1h); guardrail fisico (load_data su alt -> FileNotFoundError). Esecuzione DISABILITATA, conto flat. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
75 lines
2.8 KiB
Python
75 lines
2.8 KiB
Python
"""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()
|