# -*- coding: utf-8 -*- """ AHAD QUANT — Backtester CORRIGÉ v10-fixed (Forex Edition) ============================================= Corrections v10-fixed vs v10 : FIX A — PROFIT FACTOR mal calculé AVANT : pf = avg_win / avg_loss → donne le ratio moyen gain/perte, PAS le PF APRÈS : pf = sum(gains) / sum(|pertes|) → vrai Profit Factor standard Impact : v10 affichait 1.15 au lieu du vrai ~9.38 (identique à v11) FIX B — SHARPE biaisé par le compound sizing AVANT : calculé sur les P&L absolus en $ → les trades tardifs (~$213) pèsent 7x plus que les trades précoces (~$30), std() gonflé APRÈS : calculé sur les rendements % (pnl / position_val) → chaque trade est comparable indépendamment de la taille du compte Impact : Sharpe de 13.5 → valeur réaliste selon win rate réel FIX C — DEAD CODE supprimé (ligne eq = [...]) AVANT : eq = [balance + sum(pnl[:k]) - sum(pnl[:k]) for k in ...] calcule balance + X - X = balance pour tout k, variable jamais utilisée APRÈS : ligne supprimée FIX D — DRAWDOWN calculé sur l'equity curve RÉELLE AVANT : P&L appliqués séquentiellement trade par trade → ignore les pertes simultanées (jusqu'à MAX_POSITIONS positions ouvertes en même temps) APRÈS : equity_curve produite par run_backtest_corrected() transmise à print_results() et utilisée directement pour le calcul du DD FIX E — MARGE utilisée calculée sur les montants RÉELS AVANT : already_used_margin = n_positions × balance_actuel × RISK_PER_TRADE → approximation fausse (les positions ont été ouvertes à des niveaux de balance différents) APRÈS : open_positions stocke (expiration, marge_réelle) → la marge déjà engagée est la somme exacte des marges à l'ouverture FIX F — TP_PCT_OVERRIDE synchronisé avec config.py (v32-fixed) AVANT : TP_PCT_OVERRIDE = 0.022 (2.2%) alors que config.TAKE_PROFIT_PCT = 0.0075 (0.75%) → backtest 3× plus optimiste que le bot en production APRÈS : TP_PCT_OVERRIDE = 0.0 → utilise toujours config.TAKE_PROFIT_PCT Résultats backtest maintenant fidèles au comportement réel du bot. Nouveauté v10 vs v9 (inchangé) : Compound sizing activé, plafond $10,000/trade Héritage v9 (inchangé) : OOS strict 30%, TP 2.2%, MIN_CONFIDENCE 0.72, exclusion coins récents Usage: python backtest.py """ import json, os, pickle, time, sys sys.stdout.reconfigure(encoding="utf-8", errors="replace") from collections import defaultdict from datetime import datetime, timezone import numpy as np sys.path.insert(0, os.path.dirname(__file__)) import config from features import build_features, FEATURE_NAMES import ensemble_core as _ens_core # ── RL — MÊME filtre qu'en live (unified_brain.py), pour que le backtest # teste le système RÉEL (ML+RL), pas juste le ML seul. Import optionnel : # si stable-baselines3/gymnasium ne sont pas installés, le backtest reste # utilisable en mode ML seul (comme avant), avec un avertissement explicite. _RL_READY = False try: from rl_agent import get_rl_agent, PositionState _RL_READY = True except ImportError: PositionState = None # type: ignore[assignment,misc] # ─── Paramètres ────────────────────────────────────────────────────────────── INITIAL_BALANCE = 100.0 # Forex: compte standard $100 LEVERAGE = config.LEVERAGE # synchro config.py / .env RISK_PER_TRADE = config.RISK_PER_TRADE STOP_LOSS_PCT = config.STOP_LOSS_PCT # synchro config.py / .env TAKE_PROFIT_PCT = config.TAKE_PROFIT_PCT MIN_CONFIDENCE = config.MIN_CONFIDENCE MAX_POSITIONS = config.MAX_POSITIONS # synchro config.py / .env FEE_RATE = config.FEE_RATE # ── Slippage réaliste (FIX BUG #5) ────────────────────────────────────────── # 1.5 pips sur les majors Forex (EUR/USD, GBP/USD) — valeur broker OANDA réaliste. # Appliqué à l'entrée ET à la sortie (total : 3 pips par trade). # Pour les JPY pairs (USD/JPY), 1 pip = 0.01 donc ajuster manuellement si besoin. SLIPPAGE_PIPS = float(os.getenv("BACKTEST_SLIPPAGE_PIPS", "1.5")) PIP_SIZE = 0.0001 # taille d'1 pip pour paires non-JPY WARMUP = 50 # HISTORIQUE MINIMUM : paires avec moins de bougies que ce seuil sont exclues. # FOREX yfinance H1 : marchés fermés ~65h/semaine → ~17 000 bougies = 2.7 ans # de données réelles. Le seuil crypto (20 000) excluait TOUTES les paires Forex. # 500 bougies H1 ≈ 3 semaines — minimum statistique pour un OOS strict 15%. MIN_HISTORY_CANDLES = 500 # OUT-OF-SAMPLE : fraction des données réservée au backtest. # Les (1 - OOS_RATIO) premières bougies ont servi à l'entraînement → on ne # les touche pas. Doit correspondre à 1 - TRAIN_RATIO - VAL_RATIO de train.py # (0.70 + 0.15 = 0.85 → OOS = 0.15 au minimum ; on prend 0.30 pour la marge). OOS_RATIO = 0.15 # TEST set uniquement — 100% OOS propre # FULL HISTORY MODE : si True, le backtest tourne sur TOUTES les bougies # disponibles (pas seulement les 30% OOS finaux). Utile pour maximiser le # nombre de trades simulés et avoir une vision complète de l'historique. # ⚠️ Attention : inclut les bougies in-sample sur lesquelles le modèle a # été entraîné → les métriques seront optimistes vs. un vrai OOS test. USE_FULL_HISTORY = False # OOS strict 30% — seule mesure fiable du vrai edge live # COMPOUND SIZING : si True, la taille de position est recalculée à chaque # trade sur le solde courant (balance × RISK_PER_TRADE × LEVERAGE) au lieu # d'être fixée à INITIAL_BALANCE × RISK_PER_TRADE × LEVERAGE. # → L'effet de compounding exponentiel est activé : les gains s'accumulent. # ⚠️ En live, bien que très profitable sur backtest, le compounding amplifie # aussi les pertes. S'assurer que RISK_PER_TRADE est conservateur (≤ 3%). COMPOUND_SIZING = config.COMPOUND_ENABLED # COMPOUND_ENABLED — intérêts composés activés : position grandit avec le solde # PLAFOND DE POSITION : limite la taille max par trade en mode compound. # Sans ce plafond, à $1M de solde une position = $1M × 3% × 10x = $300K # → croissance exponentielle irréaliste (impossible en vrai sur un exchange). # $10,000 = taille max raisonnable pour un compte retail en futures. # Augmente ce plafond uniquement si tu trades avec un vrai gros capital. MAX_POSITION_VALUE = config.CAPITAL_CAP_PER_TRADE # $ max par trade — CAPITAL_CAP_PER_TRADE (compound plafonné) # OVERRIDE DU TAKE PROFIT : si > 0, remplace TAKE_PROFIT_PCT de config.py. # ⚠️ BUG CORRIGÉ : l'ancienne valeur 0.022 (2.2%) était 3× supérieure à la valeur # de production config.py (0.75%) → les résultats backtest ne correspondaient pas # au comportement réel du bot. # Valeur 0 = utilise TAKE_PROFIT_PCT de config.py / .env → cohérence garantie. TP_PCT_OVERRIDE = 0.0 # 0 = utilise config.TAKE_PROFIT_PCT (synchronisé avec la prod) # OVERRIDE DE LA CONFIANCE MINIMALE : si > 0, remplace MIN_CONFIDENCE de config.py/.env. # Analyse OOS sur 500 trades : # bucket 0.62–0.70 : WR=64.1% → destructeur de PF (20% du volume, quasi toutes les pertes nettes) # bucket 0.70–0.75 : WR=80.2% # bucket 0.75–0.80 : WR=81.7% # bucket 0.80–0.90 : WR=90.9% # Seuil 0.72 → exclut la zone 64%, garde uniquement les signals ≥80% WR. # Volume -20%, PF estimé +40–50%. MIN_CONFIDENCE_OVERRIDE = 0.72 # 0 = utilise MIN_CONFIDENCE du .env / config.py # POSITION SIZING FIXE : taille calculée sur le capital initial, pas sur le # solde courant. Évite l'explosion par compounding géométrique en backtest. # En live le bot peut utiliser un sizing dynamique, mais pour mesurer l'edge # du modèle on veut une taille constante. FIXED_POSITION_VAL = INITIAL_BALANCE * RISK_PER_TRADE * LEVERAGE # Appliquer l'override TP si activé _TP = TP_PCT_OVERRIDE if TP_PCT_OVERRIDE > 0 else TAKE_PROFIT_PCT TAKE_PROFIT_PCT = _TP # shadowed pour le reste du fichier # Appliquer l'override MIN_CONFIDENCE si activé _MC = MIN_CONFIDENCE_OVERRIDE if MIN_CONFIDENCE_OVERRIDE > 0 else MIN_CONFIDENCE MIN_CONFIDENCE = _MC # shadowed pour le reste du fichier # Nouvelle limite : jamais plus de 50 % du solde en marge simultanément MAX_MARGIN_USAGE = config.MAX_MARGIN_USAGE # fraction du solde (sans levier) — 20% max simultané # TIMEOUT : si True, les trades non résolus (ni TP ni SL) sont clôturés au # close de la bougie MAX_HOLD_CANDLES. Si False, les positions restent ouvertes # indéfiniment (non recommandé en live). TIMEOUT_ENABLED = config.TIMEOUT_ENABLED # fermeture forcée après MAX_HOLD_CANDLES bougies MAX_HOLD_CANDLES = config.MAX_HOLD_CANDLES # Risk manager simulé CIRCUIT_BREAKER_LOSSES = config.CIRCUIT_BREAKER_LOSSES # 3 CIRCUIT_BREAKER_CANDLES = config.CIRCUIT_BREAKER_COOLDOWN // 3600 # dérivé de CIRCUIT_BREAKER_COOLDOWN (en secondes → bougies 1h) MAX_DAILY_LOSS_PCT = config.MAX_DAILY_LOSS_PCT # 0.05 = 5 % # ─── Chargement modèle ─────────────────────────────────────────────────────── def load_model(): if os.path.exists(config.ENSEMBLE_MODEL_PATH): with open(config.ENSEMBLE_MODEL_PATH, "rb") as f: return pickle.load(f) with open(config.MODEL_PATH, "rb") as f: raw = pickle.load(f) m = raw["model"] if isinstance(raw, dict) else raw return {"lgbm": m, "xgb": None, "rf": None, "meta": None, "scaler": None} def load_candles(coin): path = os.path.join(config.DATA_DIR, f"{coin}_1h.json") if not os.path.exists(path): return [] with open(path) as f: return json.load(f) # ─── Étape 1 : Pré-calcul vectorisé de TOUTES les features ────────────────── def precompute_all_features(coins_data, btc_aligned): """ Appelle build_features() une seule fois par coin sur l'intégralité des données → renvoie dict coin → np.ndarray (N, 62). """ features = {} total = len(coins_data) t0 = time.perf_counter() for idx, (coin, candles) in enumerate(coins_data.items(), 1): t1 = time.perf_counter() window = [ {"o": c["o"], "h": c["h"], "l": c["l"], "c": c["c"], "v": c["v"]} for c in candles ] btc_closes = np.array([c["c"] for c in btc_aligned[-len(candles):]]) X = build_features(window, btc_closes=btc_closes) features[coin] = np.nan_to_num(X, nan=0.0, posinf=0.0, neginf=0.0) elapsed = time.perf_counter() - t1 remaining = elapsed * (total - idx) print( f" [{idx:2d}/{total}] {coin:8s} — {X.shape[0]:,} bougies, {elapsed:.1f}s" f" | reste ~{remaining:.0f}s" ) total_time = time.perf_counter() - t0 print(f"\n Pré-calcul terminé en {total_time:.1f}s ({total_time/60:.1f} min)\n") return features # ─── Étape 2 : Batch prediction pour tout le tableau ──────────────────────── def batch_predict_all(model_data, features_dict): """ Prédit la probabilité haussière pour TOUS les candles d'un coup par coin. Renvoie dict coin → np.ndarray (N,) de probabilités [0, 1]. Délègue à ensemble_core.predict_ensemble_batch() — la MÊME implémentation que celle utilisée en live (unified_brain.py) et pendant l'entraînement RL (rl_env.py). Corrige le point #1 du diagnostic (2ᵉ occurrence) : l'ancienne version ici ne gérait JAMAIS les modèles DL (TFT/TGRU) — un ensemble entraîné avec has_dl=True aurait donc été validé en backtest avec un stacking à 3 modèles, puis exécuté en live avec un stacking à 5, deux distributions de probabilités différentes pour le "même" modèle. """ probs = {} for coin, X in features_dict.items(): result = _ens_core.predict_ensemble_batch(model_data, X) probs[coin] = result[:, 0] # colonne 0 = proba_long return probs # ─── Étape 3 : Backtest corrigé ────────────────────────────────────────────── def run_backtest_corrected(probs_dict, coins_data, min_len, features_dict=None): """ Simule le bot sur l'historique avec les 3 corrections + risk manager simulé. Logique temporelle correcte : - Bougie i se FERME → features[i] connues → probs[i] calculé - On ENTRE au prix closes[i] (= clôture de la bougie i) - TP/SL résolu sur les bougies i+1, i+2, ..., i+MAX_HOLD_CANDLES `features_dict` (optionnel) : si fourni ET config.USE_RL_AGENT=true ET stable-baselines3 est installé, le signal ML de chaque bougie est filtré par le MÊME agent RL qu'en live (rl_agent.RLAgent.filter_signal()) — boost si accord, override si RL très confiant et en désaccord, fallback sinon. C'est ce qui rend ce backtest fidèle au système réel : ML+RL comme un seul système, jamais le ML testé seul puis le RL en aveugle. """ rl_agent = None if features_dict is not None and _RL_READY and getattr(config, "USE_RL_AGENT", False): try: rl_agent = get_rl_agent() if not rl_agent.is_ready(): print(" ⚠️ RL activé (USE_RL_AGENT=true) mais agent non chargé " "(rl_agent.zip absent ?) — backtest en ML seul.") rl_agent = None else: print(" ✅ Filtre RL actif pour ce backtest (ML+RL unifiés, comme en live).") except Exception as e: print(f" ⚠️ RL activé mais erreur de chargement ({e}) — backtest en ML seul.") rl_agent = None elif getattr(config, "USE_RL_AGENT", False) and not _RL_READY: print(" ⚠️ USE_RL_AGENT=true mais stable-baselines3/gymnasium non installés " "— backtest en ML seul (pip install -r requirements.txt pour corriger).") balance = INITIAL_BALANCE trades = [] equity_curve = [balance] peak_balance = balance coins = list(coins_data.keys()) closes_arr = {c: np.array([x["c"] for x in coins_data[c]]) for c in coins} highs_arr = {c: np.array([x["h"] for x in coins_data[c]]) for c in coins} lows_arr = {c: np.array([x["l"] for x in coins_data[c]]) for c in coins} # BUG 2 FIX — tracker les positions actives # open_positions[coin] = (bougie_expiration, margin_réelle_bloquée) open_positions: dict[str, tuple[int, float]] = {} # Risk manager simulé consecutive_losses = 0 circuit_breaker_until = 0 # indice de bougie daily_pnl = 0.0 current_day = -1 # jour julien de la bougie courante total_steps = min_len - WARMUP progress_interval = max(1, total_steps // 20) for i in range(WARMUP, min_len): step = i - WARMUP if step % progress_interval == 0: pct = step / total_steps * 100 print( f" [{pct:5.1f}%] bougie {i:,}/{min_len:,}" f" balance ${balance:,.2f} trades {len(trades)}" f" positions_ouvertes {len([c for c, v in open_positions.items() if v[0] > i])}" ) # ── Nettoyer les positions expirées ────────────────────────────────── open_positions = {c: v for c, v in open_positions.items() if v[0] > i} # ── Jour calendaire simulé (chaque bougie = 1h) ────────────────────── day_index = i // 24 if day_index != current_day: current_day = day_index daily_pnl = 0.0 # reset daily P&L # ── Circuit breaker simulé ─────────────────────────────────────────── if i < circuit_breaker_until: continue # ── Daily loss limit ───────────────────────────────────────────────── if balance > 0 and (daily_pnl / balance) <= -MAX_DAILY_LOSS_PCT: continue # BUG 3 FIX — compter les positions actuellement ouvertes currently_open = len(open_positions) available_slots = MAX_POSITIONS - currently_open if available_slots <= 0: continue # BUG 3 FIX — calculer la marge déjà engagée # FIX — on utilise la marge RÉELLE de chaque position ouverte (stockée à l'ouverture) # plutôt qu'une approximation basée sur le balance actuel. already_used_margin = sum(v[1] for v in open_positions.values()) available_margin = (balance if COMPOUND_SIZING else INITIAL_BALANCE) * MAX_MARGIN_USAGE - already_used_margin if available_margin <= 0: continue # ── Générer les signaux pour cette bougie ──────────────────────────── candle_signals = [] for coin in coins: # Skip si position déjà ouverte sur ce coin if coin in open_positions: continue prob = probs_dict[coin][i] if prob > MIN_CONFIDENCE: signal = "long" confidence = prob elif prob < (1 - MIN_CONFIDENCE): signal = "short" confidence = 1 - prob else: continue # ── Filtre RL — EXACTEMENT la même règle qu'en live # (unified_brain.UnifiedBrain.decide()) : uniquement sur signal # ML non-neutre, jamais de re-calibration silencieuse ici. ── if rl_agent is not None: try: last_feat = features_dict[coin][i] pos_state = PositionState(in_position=False, direction=0, balance_ratio=1.0) rl_signal, rl_confidence, _rl_action = rl_agent.filter_signal( ml_signal=signal, ml_confidence=confidence, features=last_feat, position_state=pos_state, ) signal, confidence = rl_signal, rl_confidence if signal == "neutral": continue except Exception: pass # fallback silencieux sur le signal ML, comme en live # BUG 1 FIX — entry au close de la bougie i (prix connu à la clôture) entry = closes_arr[coin][i] if entry <= 0: continue # FIX BUG #5 — Slippage réaliste à l'entrée (1.5 pips) slip = SLIPPAGE_PIPS * PIP_SIZE if signal == "long": entry = entry * (1 + slip) # achat : prix monte légèrement else: entry = entry * (1 - slip) # vente : prix descend légèrement if signal == "long": sl = entry * (1 - STOP_LOSS_PCT) tp = entry * (1 + TAKE_PROFIT_PCT) else: sl = entry * (1 + STOP_LOSS_PCT) tp = entry * (1 - TAKE_PROFIT_PCT) # BUG 1 FIX — future commence à i+1 (bougie SUIVANTE, pas la même) fut_start = i + 1 fut_end = min(i + 1 + MAX_HOLD_CANDLES, min_len) if fut_start >= min_len: continue # plus de données pour résoudre ce trade future_h = highs_arr[coin][fut_start:fut_end] future_l = lows_arr[coin][fut_start:fut_end] future_c = closes_arr[coin][fut_start:fut_end] if signal == "long": sl_hit = np.where(future_l <= sl)[0] tp_hit = np.where(future_h >= tp)[0] else: sl_hit = np.where(future_h >= sl)[0] tp_hit = np.where(future_l <= tp)[0] sl_idx = sl_hit[0] if len(sl_hit) > 0 else 999 tp_idx = tp_hit[0] if len(tp_hit) > 0 else 999 if sl_idx == tp_idx == 999: if not TIMEOUT_ENABLED: continue # position ignorée si timeout désactivé result = "timeout" exit_price = future_c[-1] if len(future_c) > 0 else entry hold_dur = len(future_c) elif tp_idx <= sl_idx: result = "tp" exit_price = tp hold_dur = tp_idx + 1 else: result = "sl" exit_price = sl hold_dur = sl_idx + 1 # FIX BUG #5 — Slippage réaliste à la sortie (1.5 pips) if signal == "long": exit_price = exit_price * (1 - slip) # vente : prix descend else: exit_price = exit_price * (1 + slip) # rachat : prix monte if signal == "long": pnl_pct = (exit_price - entry) / entry else: pnl_pct = (entry - exit_price) / entry # COMPOUND_SIZING : taille recalculée sur le solde courant, plafonnée if COMPOUND_SIZING: position_val = min( max(balance * RISK_PER_TRADE * LEVERAGE, 0.0), MAX_POSITION_VALUE ) else: position_val = FIXED_POSITION_VAL fee = position_val * FEE_RATE * 2 net_pnl = position_val * pnl_pct - fee candle_signals.append({ "coin": coin, "signal": signal, "confidence": float(confidence), "result": result, "entry": float(entry), "exit": float(exit_price), "pnl": float(net_pnl), "position_val": float(position_val), "hold_candles": int(hold_dur), "candle_idx": i, }) # ── Sélectionner les meilleurs signaux dans la limite des slots ─────── candle_signals.sort(key=lambda x: x["confidence"], reverse=True) margin_used_this_step = 0.0 for t in candle_signals: if len(open_positions) >= MAX_POSITIONS: break margin_needed_check = (balance if COMPOUND_SIZING else INITIAL_BALANCE) * RISK_PER_TRADE if margin_used_this_step + margin_needed_check > available_margin: break # Enregistrer la position comme ouverte jusqu'à sa bougie de clôture # FIX — on stocke (expiration, margin_réelle) pour un calcul de marge exact margin_this_trade = (balance if COMPOUND_SIZING else INITIAL_BALANCE) * RISK_PER_TRADE close_at = i + 1 + t["hold_candles"] open_positions[t["coin"]] = (close_at, margin_this_trade) margin_used_this_step += margin_this_trade # Appliquer le P&L balance = max(0.0, balance + t["pnl"]) trades.append(t) equity_curve.append(balance) peak_balance = max(peak_balance, balance) # ── Risk manager simulé ────────────────────────────────────────── daily_pnl += t["pnl"] if t["pnl"] < 0: consecutive_losses += 1 if consecutive_losses >= CIRCUIT_BREAKER_LOSSES: circuit_breaker_until = i + CIRCUIT_BREAKER_CANDLES consecutive_losses = 0 pass # circuit breaker silencieux (trop verbeux en console) else: consecutive_losses = 0 if balance == 0.0: print(f" 💀 Balance = 0 à la bougie {i} — simulation arrêtée") return trades, equity_curve, (rl_agent is not None) return trades, equity_curve, (rl_agent is not None) # ─── Affichage des résultats ────────────────────────────────────────────────── def print_results(trades, balance, equity_curve=None): print("\n" + "=" * 62) print(" RÉSULTATS DU BACKTEST (v10-fixed — OOS STRICT + COMPOUND + TP 2.2% + CONF 0.72)") print("=" * 62) if not trades: print(" Aucun trade généré.") return wins = [t for t in trades if t["pnl"] > 0] loss = [t for t in trades if t["pnl"] <= 0] tps = [t for t in trades if t["result"] == "tp"] sls = [t for t in trades if t["result"] == "sl"] tmos = [t for t in trades if t["result"] == "timeout"] longs = [t for t in trades if t["signal"] == "long"] shts = [t for t in trades if t["signal"] == "short"] total_pnl = balance - INITIAL_BALANCE pnl_pct = total_pnl / INITIAL_BALANCE * 100 win_rate = len(wins) / len(trades) * 100 if trades else 0 # Drawdown max — calculé sur l'equity curve RÉELLE du backtest # FIX — l'ancienne méthode appliquait les trades séquentiellement alors que # plusieurs positions sont ouvertes simultanément : le vrai drawdown peut # être plus élevé. On utilise l'equity_curve produite par run_backtest_corrected. eq_for_dd = equity_curve if equity_curve and len(equity_curve) > 1 else None if eq_for_dd: peak = eq_for_dd[0] max_dd = 0.0 for v in eq_for_dd: peak = max(peak, v) dd = (peak - v) / peak * 100 if peak > 0 else 0 max_dd = max(max_dd, dd) else: # Fallback séquentiel si equity_curve non fournie peak = INITIAL_BALANCE max_dd = 0.0 running = INITIAL_BALANCE for t in trades: running += t["pnl"] peak = max(peak, running) dd = (peak - running) / peak * 100 if peak > 0 else 0 max_dd = max(max_dd, dd) avg_win = np.mean([t["pnl"] for t in wins]) if wins else 0 avg_loss = np.mean([t["pnl"] for t in loss]) if loss else 0 # FIX — vrai Profit Factor = somme des gains / somme des pertes # (l'ancienne formule avg_win/avg_loss donnait le ratio moyen, pas le PF) gross_profit = sum(t["pnl"] for t in wins) gross_loss = abs(sum(t["pnl"] for t in loss)) pf = gross_profit / gross_loss if gross_loss > 0 else float("inf") best = max(trades, key=lambda x: x["pnl"]) worst = min(trades, key=lambda x: x["pnl"]) avg_hold = np.mean([t["hold_candles"] for t in trades]) if trades else 0 # Sharpe (calculé sur les rendements % par trade, pas sur le P&L absolu) # FIX — le P&L absolu est biaisé par le compound : les trades tardifs valent # 1000x plus qu'au départ, ce qui gonfle artificiellement le Sharpe. # On normalise chaque trade par la taille de position pour obtenir # un rendement % comparable quelle que soit la taille du compte. # Annualisation : ~24 trades/jour × 252 jours = 6048 périodes/an. ret_series = np.array([ t["pnl"] / max(t.get("position_val", FIXED_POSITION_VAL), 1e-9) for t in trades ]) sharpe = (ret_series.mean() / ret_series.std() * np.sqrt(6048)) \ if ret_series.std() > 0 else 0 # Top/flop coins coin_pnl: dict[str, float] = {} for t in trades: coin_pnl[t["coin"]] = coin_pnl.get(t["coin"], 0) + t["pnl"] top_coins = sorted(coin_pnl.items(), key=lambda x: x[1], reverse=True) flop_coins = sorted(coin_pnl.items(), key=lambda x: x[1]) print(f"\n 💰 Balance initiale : ${INITIAL_BALANCE:>12,.2f}") print(f" 💰 Balance finale : ${balance:>12,.2f}") print(f" 📈 PnL total : ${total_pnl:>+12,.2f} ({pnl_pct:+.1f}%)") print(f" 📉 Drawdown max : {max_dd:.1f}%") print(f" 📐 Sharpe (approx.) : {sharpe:.2f}") print(f" ⚖️ Profit Factor : {pf:.2f}") # Compound sizing info if COMPOUND_SIZING: cap_balance = MAX_POSITION_VALUE / (RISK_PER_TRADE * LEVERAGE) print(f" 🔄 Compound : ON — plafond atteint à ~${cap_balance:,.0f} de solde") print(f"\n 📊 Trades total : {len(trades):,}") print(f" ✅ Gagnants : {len(wins):,} ({win_rate:.1f}%)") print(f" ❌ Perdants : {len(loss):,} ({100-win_rate:.1f}%)") print(f" 🎯 TP atteints : {len(tps):,}") print(f" 🛑 SL atteints : {len(sls):,}") print(f" ⏱️ Timeouts : {len(tmos):,}") print(f" 📗 Longs / 📕 Shorts : {len(longs):,} / {len(shts):,}") print(f" ⌛ Durée moy. trade : {avg_hold:.1f} bougies") print(f"\n 💵 Gain moyen : ${avg_win:>+10,.2f}") print(f" 💵 Perte moyenne : ${avg_loss:>+10,.2f}") print(f" 🏆 Meilleur trade : ${best['pnl']:>+10,.2f} ({best['coin']} {best['signal'].upper()})") print(f" 💀 Pire trade : ${worst['pnl']:>+10,.2f} ({worst['coin']} {worst['signal'].upper()})") print(f"\n 🏅 Top 5 coins:") max_coin_pnl = abs(top_coins[0][1]) if top_coins else 1 for coin, pnl in top_coins[:5]: bar = "█" * int(max(0, pnl) / max(1, max_coin_pnl) * 20) print(f" {coin:8s} ${pnl:>+10,.2f} {bar}") print(f"\n 💀 Flop 5 coins:") for coin, pnl in flop_coins[:5]: bar = "█" * int(max(0, -pnl) / max(1, max_coin_pnl) * 20) print(f" {coin:8s} ${pnl:>+10,.2f} {bar}") print("\n" + "=" * 62) # ── Diagnostic ────────────────────────────────────────────────────────── print("\n DIAGNOSTIC :") if pnl_pct > 50: print(" ✅ PnL > +50% — Fort sur historique, vérifie en paper") elif pnl_pct > 20: print(" ✅ PnL > +20% — Excellent") elif pnl_pct > 5: print(" 🟡 PnL +5% à +20% — Acceptable, passe en paper trading") elif pnl_pct > 0: print(" 🟠 PnL faiblement + — Augmente MIN_CONFIDENCE ou réduis levier") else: print(" ❌ PnL négatif — Ré-entraîne avec OPTUNA_TRIALS=50 dans train.py") if max_dd < 15: print(" ✅ DD < 15% — Très sûr pour 5x levier") elif max_dd < 25: print(" 🟡 DD 15-25% — Acceptable, surveille en paper trading") else: print(" ❌ DD > 25% — Réduis RISK_PER_TRADE ou LEVERAGE dans .env") if win_rate > 52: print(" ✅ Win rate > 52% — Excellent avec R:R 1:2") elif win_rate > 45: print(" 🟡 Win rate 45-52% — OK, le R:R compense") else: print(" ❌ Win rate < 45% — Augmente MIN_CONFIDENCE à 0.72+") if sharpe > 1.5: print(" ✅ Sharpe > 1.5 — Très bon rapport risque/rendement") elif sharpe > 1.0: print(" 🟡 Sharpe 1.0-1.5 — Correct") else: print(" 🟠 Sharpe < 1.0 — Stratégie volatile, réduis le levier") print("=" * 62) # ── Equity curve ASCII dans le terminal ────────────────────────────────── W = 62 print("\n" + "=" * W) print(" COURBE D\'ÉQUITÉ (terminal)") print("=" * W) # Rééchantillonner l'equity curve sur 50 points max # Reconstruire l'equity curve proprement # FIX — suppression du dead code : l'ancienne ligne calculait balance + X - X = balance running = INITIAL_BALANCE eq_vals = [running] for t in trades: running = max(0.0, running + t["pnl"]) eq_vals.append(running) n_pts = min(50, len(eq_vals)) step = max(1, len(eq_vals) // n_pts) sample = eq_vals[::step] if eq_vals[-1] not in sample: sample.append(eq_vals[-1]) eq_min = min(sample) eq_max = max(sample) rows = 8 print() for row in range(rows, -1, -1): thresh = eq_min + (eq_max - eq_min) * row / rows if row == rows: label = f"${eq_max:>10,.0f} │" elif row == 0: label = f"${eq_min:>10,.0f} │" else: label = f"{'':>11} │" line = "" for v in sample: norm = (v - eq_min) / max(eq_max - eq_min, 1) filled = norm * rows >= row line += "█" if filled else " " print(f" {label}{line}") print(f" {'':>11} └{'─'*len(sample)}") print(f" {'':>12} début{'':>{max(0,len(sample)-12)}}fin") # ── Distribution des trades par résultat ───────────────────────────────── print("\n" + "=" * W) print(" DISTRIBUTION DES TRADES") print("=" * W) total_t = len(trades) tp_pct_t = len(tps) / total_t * 100 if total_t else 0 sl_pct_t = len(sls) / total_t * 100 if total_t else 0 tmo_pct_t = len(tmos) / total_t * 100 if total_t else 0 bar_w = 30 def pct_bar(pct, char="█"): n = int(pct / 100 * bar_w) return char * n + "░" * (bar_w - n) print(f" 🎯 TP {pct_bar(tp_pct_t)} {len(tps):>6,} ({tp_pct_t:5.1f}%)") print(f" 🛑 SL {pct_bar(sl_pct_t)} {len(sls):>6,} ({sl_pct_t:5.1f}%)") print(f" ⏱ Timeout {pct_bar(tmo_pct_t)} {len(tmos):>6,} ({tmo_pct_t:5.1f}%)") # ── PnL par coin (barres horizontales) ─────────────────────────────────── print("\n" + "=" * W) print(" PnL PAR COIN (tous les coins)") print("=" * W) all_coins = sorted(coin_pnl.items(), key=lambda x: x[1], reverse=True) max_abs = max(abs(p) for _, p in all_coins) if all_coins else 1 for coin, pnl in all_coins: bar_len = int(abs(pnl) / max_abs * 20) sign = "+" if pnl >= 0 else "-" bar = ("█" if pnl >= 0 else "░") * bar_len print(f" {coin:6s} {sign}${abs(pnl):>10,.0f} {bar}") print("\n" + "=" * W + "\n") # ─── Sauvegarde des résultats ───────────────────────────────────────────────── def save_results(trades, balance, equity_curve, run_timestamp=None, rl_filter_active=False): """ Sauvegarde les résultats du backtest dans backtest_results.json. ⚠️ ÉCRASE TOUJOURS le fichier précédent — un seul fichier de résultats, mis à jour à chaque nouveau backtest (pas d'accumulation). Format identique à l'ancien backtest_results.json pour la compatibilité avec dashboard.py / web_ui.py. """ if not trades: print(" [SAVE] Aucun trade — fichier non écrasé.") return wins = [t for t in trades if t["pnl"] > 0] loss = [t for t in trades if t["pnl"] <= 0] tps = [t for t in trades if t["result"] == "tp"] sls = [t for t in trades if t["result"] == "sl"] tmos = [t for t in trades if t["result"] == "timeout"] total_pnl = balance - INITIAL_BALANCE pnl_pct = total_pnl / INITIAL_BALANCE * 100 win_rate = len(wins) / len(trades) * 100 if trades else 0 # Drawdown sur equity curve réelle eq = equity_curve if equity_curve and len(equity_curve) > 1 else None if eq: peak, max_dd = eq[0], 0.0 for v in eq: peak = max(peak, v) dd = (peak - v) / peak * 100 if peak > 0 else 0 max_dd = max(max_dd, dd) else: peak, max_dd, running = INITIAL_BALANCE, 0.0, INITIAL_BALANCE for t in trades: running += t["pnl"] peak = max(peak, running) dd = (peak - running) / peak * 100 if peak > 0 else 0 max_dd = max(max_dd, dd) gross_profit = sum(t["pnl"] for t in wins) gross_loss = abs(sum(t["pnl"] for t in loss)) pf = gross_profit / gross_loss if gross_loss > 0 else float("inf") ret_series = np.array([ t["pnl"] / max(t.get("position_val", FIXED_POSITION_VAL), 1e-9) for t in trades ]) sharpe = (ret_series.mean() / ret_series.std() * np.sqrt(6048)) \ if ret_series.std() > 0 else 0 avg_hold = float(np.mean([t["hold_candles"] for t in trades])) if trades else 0 coin_pnl: dict[str, float] = {} for t in trades: coin_pnl[t["coin"]] = coin_pnl.get(t["coin"], 0) + t["pnl"] result = { # ── Metadata ────────────────────────────────────────────────────── "backtest_version": "v10-fixed", "run_timestamp": run_timestamp or datetime.now(timezone.utc).isoformat(), "rl_filter_active": bool(rl_filter_active), "system": "ML+RL unifié" if rl_filter_active else "ML seul (RL indisponible)", "corrections": [ "profit_factor_fixed", "sharpe_pct_based", "dead_code_removed", "drawdown_real_equity", "margin_real_values", ], # ── Paramètres utilisés ─────────────────────────────────────────── "min_history_candles": MIN_HISTORY_CANDLES, "oos_ratio": OOS_RATIO, "use_full_history": USE_FULL_HISTORY, "compound_sizing": COMPOUND_SIZING, "tp_pct_used": TAKE_PROFIT_PCT, "sl_pct_used": STOP_LOSS_PCT, "min_confidence_used": MIN_CONFIDENCE, "max_position_value": MAX_POSITION_VALUE, "fixed_position_val": FIXED_POSITION_VAL, # ── Métriques principales ───────────────────────────────────────── "balance_initial": INITIAL_BALANCE, "balance_final": float(balance), "total_pnl": float(total_pnl), "pnl_pct": float(pnl_pct), "win_rate": float(win_rate), "max_drawdown": float(max_dd), "sharpe": float(sharpe), "profit_factor": float(pf), # ── Compteurs trades ────────────────────────────────────────────── "total_trades": len(trades), "tp_count": len(tps), "sl_count": len(sls), "timeout_count": len(tmos), "avg_hold_candles": float(avg_hold), # ── Config active ───────────────────────────────────────────────── "config": { "leverage": LEVERAGE, "risk_per_trade": RISK_PER_TRADE, "stop_loss_pct": STOP_LOSS_PCT, "take_profit_pct": TAKE_PROFIT_PCT, "min_confidence": MIN_CONFIDENCE, "max_positions": MAX_POSITIONS, "max_margin_usage": MAX_MARGIN_USAGE, }, # ── PnL par coin ────────────────────────────────────────────────── "coin_pnl": {k: float(v) for k, v in coin_pnl.items()}, # ── 500 derniers trades pour le dashboard ───────────────────────── "trades_last_500": trades[-500:], # ── Equity curve rééchantillonnée sur 500 points max ────────────── "equity_curve_sampled": ( equity_curve[:: max(1, len(equity_curve) // 500)] if equity_curve else [] ), } out_path = os.path.join(os.path.dirname(__file__), "backtest_results.json") # Écriture atomique : on écrit d'abord dans un fichier temporaire, # puis on le renomme → évite un fichier corrompu en cas d'interruption. tmp_path = out_path + ".tmp" with open(tmp_path, "w", encoding="utf-8") as f: json.dump(result, f, indent=2, ensure_ascii=False) os.replace(tmp_path, out_path) # atomique sur Linux/Windows size_kb = os.path.getsize(out_path) / 1024 print(f"\n 💾 Résultats sauvegardés → backtest_results.json ({size_kb:.0f} KB)") print(f" Run : {result['run_timestamp']}") print(f" {len(trades):,} trades | balance ${balance:,.2f} | PnL {pnl_pct:+.1f}%") # ─── Main ───────────────────────────────────────────────────────────────────── def main(): t_global = time.perf_counter() print("=" * 62) print(" AHAD QUANT — Backtester v10-fixed — OOS STRICT + COMPOUND + R:R 1:1.47") print(" (OOS strict 30% + compound sizing + TP 2.2% + confiance 0.72)") print("=" * 62) print( f"\n Config : levier {LEVERAGE}x | SL {STOP_LOSS_PCT*100:.1f}%" f" | TP {TAKE_PROFIT_PCT*100:.1f}% | confiance > {MIN_CONFIDENCE*100:.0f}%" f" | margin_max {MAX_MARGIN_USAGE*100:.0f}%" f" | compound {'ON 🚀' if COMPOUND_SIZING else 'OFF'}" f" | cap ${MAX_POSITION_VALUE:,.0f}/trade" ) # ── Chargement ──────────────────────────────────────────────────────────── print("\n[1/4] Chargement des données...") # Paire de référence pour la corrélation (EURUSD = paire dominante Forex) _ref_pair = "EURUSD" btc_candles = load_candles(_ref_pair) if not btc_candles: print(f" [ERREUR] {_ref_pair}_1h.json introuvable. Lance download_data.py d'abord.") return coins_data: dict[str, list] = {} excluded = [] for coin in config.COINS: c = load_candles(coin) if len(c) >= MIN_HISTORY_CANDLES: coins_data[coin] = c else: excluded.append(f"{coin}({len(c)}b)") if excluded: print(f" ⚠️ Paires exclues (historique insuffisant < {MIN_HISTORY_CANDLES:,} bougies) :") print(f" {', '.join(excluded)}") if not coins_data: print(" [ERREUR] Aucune donnée trouvée dans data/. Lance download_data.py (Forex).") return # Aligner toutes les séries sur la même longueur min_len = min(len(c) for c in coins_data.values()) min_len = min(min_len, len(btc_candles)) for coin in coins_data: coins_data[coin] = coins_data[coin][-min_len:] btc_aligned = btc_candles[-min_len:] # ── Découpage Out-Of-Sample (ou historique complet) ───────────────────── if USE_FULL_HISTORY: # MODE FULL HISTORY : toutes les bougies disponibles sont utilisées. # Aucune donnée n'est exclue — le backtest maximise le nombre de trades. # ⚠️ Les bougies in-sample (vues à l'entraînement) sont incluses : # les métriques sont donc optimistes par rapport au vrai live. oos_start = 0 print(f" {len(coins_data)} coins | {min_len:,} bougies total " f"| MODE FULL HISTORY (toutes les bougies) " f"| OOS strict désactivé") else: # MODE OOS STRICT : on ne garde QUE les dernières OOS_RATIO bougies. # train.py : TRAIN=70%, VAL=15%, TEST=15% → OOS_RATIO=0.30 englobe # entièrement le set de test + une marge de sécurité supplémentaire. oos_start = int(min_len * (1 - OOS_RATIO)) for coin in coins_data: coins_data[coin] = coins_data[coin][oos_start:] btc_aligned = btc_aligned[oos_start:] min_len = min(len(c) for c in coins_data.values()) n_total_candles = min_len + oos_start print(f" {len(coins_data)} coins | {n_total_candles:,} bougies total " f"| OOS : {min_len:,} bougies ({OOS_RATIO*100:.0f}% finaux) " f"| In-sample ignoré : {oos_start:,} bougies") # ── Pré-calcul features ─────────────────────────────────────────────────── print(f"\n[2/4] Pré-calcul des features (1 appel / coin)...") print(f" Estimation : ~{len(coins_data) * 17:.0f}s ({len(coins_data) * 17/60:.1f} min)\n") features_dict = precompute_all_features(coins_data, btc_aligned) # ── Batch prediction ────────────────────────────────────────────────────── print("[3/4] Prédiction batch (tout le tableau d'un coup)...") t_pred = time.perf_counter() model_data = load_model() probs_dict = batch_predict_all(model_data, features_dict) print(f" Terminé en {time.perf_counter()-t_pred:.1f}s\n") # ── Backtest ────────────────────────────────────────────────────────────── print(f"[4/4] Simulation backtest corrigée ({min_len - WARMUP:,} steps)...\n") trades, equity_curve, rl_filter_active = run_backtest_corrected( probs_dict, coins_data, min_len, features_dict=features_dict ) # ── Résultats ───────────────────────────────────────────────────────────── final_balance = equity_curve[-1] if equity_curve else INITIAL_BALANCE run_ts = datetime.now(timezone.utc).isoformat() print_results(trades, final_balance, equity_curve) # ── Sauvegarde — écrase toujours le backtest précédent ──────────────── save_results(trades, final_balance, equity_curve, run_timestamp=run_ts, rl_filter_active=rl_filter_active) total_time = time.perf_counter() - t_global print(f" Temps total : {total_time:.0f}s ({total_time/60:.1f} min)") if __name__ == "__main__": main()