gavindiaz/ahad-quant
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
ahad-quant/backtest.py
T

973 lines
46 KiB
Python
Raw Permalink Normal View History

Initial commit - AHAD QUANT v1
2026-06-25 14:00:20 +03:00
# -*- 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()
Reference in New Issue Copy Permalink