""" Backtest strategia RETTANGOLO 2H - BREAKOUT + RETEST. Scarica daily (per range) + 2H (per candele), applica logica breakout+retest. """ import csv import json import time import urllib.request import urllib.parse from datetime import datetime, timezone from rettangolo_strategy import compute_signal ASSETS = ["SOLUSDT", "BTCUSDT", "ETHUSDT"] DAYS_BACK = 180 SL_BUFFER_PCT = 0.01 OUT_DIR = r"G:\AI TRADING ENGINE\live_deploy\bt_rettangolo_2h" def fetch_klines(symbol, interval, days): end_ms = int(time.time() * 1000) start_ms = end_ms - days * 24 * 3600 * 1000 params = {"category": "linear", "symbol": symbol, "interval": interval, "start": start_ms, "end": end_ms, "limit": 200} qs = urllib.parse.urlencode(params) url = f"https://api.bybit.com/v5/market/kline?{qs}" with urllib.request.urlopen(url, timeout=15) as r: data = json.loads(r.read().decode("utf-8")) rows = data.get("result", {}).get("list", []) rows.sort(key=lambda x: int(x[0])) klines = [] for r in rows: klines.append({ "ts": int(r[0]), "open": float(r[1]), "high": float(r[2]), "low": float(r[3]), "close": float(r[4]), "volume": float(r[5]) }) return klines def backtest_symbol(symbol): daily = fetch_klines(symbol, "D", DAYS_BACK) intraday = fetch_klines(symbol, "120", DAYS_BACK) if len(daily) < 5 or len(intraday) < 50: return None trades = [] # mappa intraday per ts intra_by_ts = {k["ts"]: k for k in intraday} # per ogni candela daily (esclusa la prima), usa il range daily precedente for d in range(1, len(daily)): prev_day = daily[d - 1] cur_day = daily[d] rng_top = prev_day["high"] rng_bot = prev_day["low"] rng_mid = (rng_top + rng_bot) / 2 if rng_top <= rng_bot: continue # finestra intraday: dalla apertura di cur_day alla sua chiusura day_start = cur_day["ts"] day_end = day_start + 24 * 3600 * 1000 day_intra = [k for k in intraday if day_start <= k["ts"] < day_end] if len(day_intra) < 3: continue for i in range(1, len(day_intra)): sig = compute_signal(prev_day, day_intra, i, SL_BUFFER_PCT) if not sig: continue entry = sig["entry"] sl = sig["sl"] tp = sig["tp"] outcome = "OPEN" exit_price = day_intra[-1]["close"] for j in range(i + 1, len(day_intra)): bar = day_intra[j] if sig["signal"] == "LONG": if bar["low"] <= sl: outcome = "SL"; exit_price = sl; break if bar["high"] >= tp: outcome = "TP"; exit_price = tp; break else: if bar["high"] >= sl: outcome = "SL"; exit_price = sl; break if bar["low"] <= tp: outcome = "TP"; exit_price = tp; break if sig["signal"] == "LONG": pnl_pct = (exit_price - entry) / entry * 100 else: pnl_pct = (entry - exit_price) / entry * 100 trades.append({ "date": day_intra[i]["ts"], "signal": sig["signal"], "entry": entry, "sl": sl, "tp": tp, "outcome": outcome, "exit": exit_price, "pnl_pct": round(pnl_pct, 2), }) return trades def summarize(symbol, trades): if not trades: return None n = len(trades) wins = sum(1 for t in trades if t["outcome"] == "TP") losses = sum(1 for t in trades if t["outcome"] == "SL") opens = sum(1 for t in trades if t["outcome"] == "OPEN") wr = wins / n * 100 avg_win = sum(t["pnl_pct"] for t in trades if t["outcome"] == "TP") / max(wins, 1) avg_loss = sum(t["pnl_pct"] for t in trades if t["outcome"] == "SL") / max(losses, 1) total_pnl = sum(t["pnl_pct"] for t in trades) longs = sum(1 for t in trades if t["signal"] == "LONG") shorts = n - longs cum = 0 peak = 0 max_dd = 0 for t in trades: cum += t["pnl_pct"] peak = max(peak, cum) dd = peak - cum max_dd = max(max_dd, dd) return { "symbol": symbol, "n_trades": n, "wins": wins, "losses": losses, "open_at_end": opens, "win_rate_pct": round(wr, 1), "avg_win_pct": round(avg_win, 2), "avg_loss_pct": round(avg_loss, 2), "total_pnl_pct": round(total_pnl, 2), "max_dd_pct": round(max_dd, 2), "longs": longs, "shorts": shorts, } import os os.makedirs(OUT_DIR, exist_ok=True) print(f"Backtest RETTANGOLO 2H - BREAKOUT + RETEST - ultimi {DAYS_BACK} giorni") print(f"Asset: {', '.join(ASSETS)}\n") all_results = [] for sym in ASSETS: print(f"[{sym}] scarico + backtest...") trades = backtest_symbol(sym) if trades is None: print(f" pochi dati, skip\n") continue # salva csv csv_path = os.path.join(OUT_DIR, f"{sym}_trades.csv") with open(csv_path, "w", newline="", encoding="utf-8") as f: w = csv.DictWriter(f, fieldnames=list(trades[0].keys())) w.writeheader() w.writerows(trades) summary = summarize(sym, trades) all_results.append(summary) if summary: print(f" trades: {summary['n_trades']} | WR: {summary['win_rate_pct']}% | avg W: {summary['avg_win_pct']}% | avg L: {summary['avg_loss_pct']}% | tot PnL: {summary['total_pnl_pct']}% | max DD: {summary['max_dd_pct']}%") print() # === REPORT === print("=" * 90) print(f"REPORT FINALE - Strategia Rettangolo 2H (Breakout+Retest) - {DAYS_BACK} giorni") print("=" * 90) print(f"{'Asset':<12} {'N':>5} {'W':>4} {'L':>4} {'Open':>4} {'WR%':>6} {'avgW%':>7} {'avgL%':>7} {'totPnL%':>9} {'maxDD%':>7} {'L/S':>6}") print("-" * 90) tot_trades = 0 tot_wins = 0 tot_losses = 0 tot_pnl = 0 worst_dd = 0 for r in all_results: print(f"{r['symbol']:<12} {r['n_trades']:>5} {r['wins']:>4} {r['losses']:>4} {r['open_at_end']:>4} {r['win_rate_pct']:>6} {r['avg_win_pct']:>7} {r['avg_loss_pct']:>7} {r['total_pnl_pct']:>9} {r['max_dd_pct']:>7} {r['longs']}/{r['shorts']:>4}") tot_trades += r['n_trades'] tot_wins += r['wins'] tot_losses += r['losses'] tot_pnl += r['total_pnl_pct'] worst_dd = max(worst_dd, r['max_dd_pct']) print("-" * 90) if tot_trades > 0: wr = tot_wins / tot_trades * 100 print(f"{'TOTALE':<12} {tot_trades:>5} {tot_wins:>4} {tot_losses:>4} {'':>4} {wr:>6.1f} {'':>7} {'':>7} {tot_pnl:>9.2f} {worst_dd:>7}") print(f"\nReport salvato in: {OUT_DIR}")