import pandas as pd import numpy as np def run_backtest( prices_a: pd.Series, prices_b: pd.Series, signals: pd.DataFrame, beta: float, transaction_cost: float = 0.001, ) -> dict: """ Simule les trades de pairs trading et calcule les performances. Pour chaque jour en position : - Long spread = acheter A, shorter B (pondéré par beta) - P&L quotidien = variation du spread * position - Frais de transaction appliqués à chaque entrée/sortie Parameters ---------- prices_a : pd.Series Prix du premier actif. prices_b : pd.Series Prix du second actif. signals : pd.DataFrame DataFrame issu de generate_signals (colonnes: zscore, signal, trade_type). beta : float Coefficient de hedge. transaction_cost : float Coût de transaction en proportion (défaut : 0.001 = 0.1%). Returns ------- dict Métriques de performance + courbe d'equity + log des trades. """ # Calcul du spread spread = prices_a - beta * prices_b # Variation quotidienne du spread spread_diff = spread.diff().fillna(0) # Position du signal alignée sur l'index des prix position = signals["signal"].reindex(spread.index).fillna(0) # P&L quotidien = position de la veille * variation du spread du jour # (on entre en fin de journée, le P&L commence le lendemain) position_decalee = position.shift(1).fillna(0) pnl_quotidien = position_decalee * spread_diff # Frais de transaction : appliqués quand la position change changement_position = position.diff().fillna(0) # Le coût est proportionnel à la valeur notionnelle (prix A + beta * prix B) valeur_notionnelle = prices_a + beta * prices_b frais = abs(changement_position) * valeur_notionnelle * transaction_cost pnl_quotidien = pnl_quotidien - frais # Courbe d'equity (cumul du P&L) equity = pnl_quotidien.cumsum() # --- Extraction du log des trades --- liste_trades = _extraire_trades(signals, spread) # --- Calcul des métriques --- rendement_total = float(equity.iloc[-1]) nb_jours = len(equity) nb_annees = nb_jours / 252 # Rendement annualisé (basé sur le P&L absolu rapporté à la valeur notionnelle moyenne) valeur_notionnelle_moyenne = float(valeur_notionnelle.mean()) rendement_total_pct = rendement_total / valeur_notionnelle_moyenne * 100 rendement_annualise = rendement_total_pct / nb_annees if nb_annees > 0 else 0.0 # Sharpe ratio (annualisé) rendements_quotidiens = pnl_quotidien / valeur_notionnelle_moyenne sharpe_ratio = _calculer_sharpe(rendements_quotidiens) # Max drawdown max_drawdown = _calculer_max_drawdown(equity) # Statistiques des trades pnl_par_trade = [t["pnl"] for t in liste_trades] trades_gagnants = [p for p in pnl_par_trade if p > 0] trades_perdants = [p for p in pnl_par_trade if p <= 0] nb_trades = len(liste_trades) win_rate = len(trades_gagnants) / nb_trades * 100 if nb_trades > 0 else 0.0 durees = [t["duree_jours"] for t in liste_trades] duree_moyenne = np.mean(durees) if durees else 0.0 gains_bruts = sum(trades_gagnants) if trades_gagnants else 0.0 pertes_brutes = abs(sum(trades_perdants)) if trades_perdants else 0.0 profit_factor = gains_bruts / pertes_brutes if pertes_brutes > 0 else float("inf") return { "total_return": round(rendement_total, 2), "total_return_pct": round(rendement_total_pct, 2), "annualized_return_pct": round(rendement_annualise, 2), "sharpe_ratio": round(sharpe_ratio, 2), "max_drawdown": round(max_drawdown, 2), "win_rate": round(win_rate, 1), "num_trades": nb_trades, "avg_trade_duration": round(duree_moyenne, 1), "profit_factor": round(profit_factor, 2), "equity_curve": equity, "trade_log": liste_trades, } def _calculer_sharpe(rendements_quotidiens: pd.Series, taux_sans_risque: float = 0.0) -> float: """ Calcule le ratio de Sharpe annualisé. sharpe = (rendement_moyen - taux_sans_risque) / volatilité * sqrt(252) """ rendement_moyen = rendements_quotidiens.mean() volatilite = rendements_quotidiens.std() if volatilite == 0 or np.isnan(volatilite): return 0.0 sharpe = (rendement_moyen - taux_sans_risque / 252) / volatilite * np.sqrt(252) return float(sharpe) def _calculer_max_drawdown(equity: pd.Series) -> float: """ Calcule le max drawdown en valeur absolue. max_drawdown = pire baisse depuis un pic de la courbe d'equity. """ pic = equity.cummax() drawdown = equity - pic return float(drawdown.min()) def _extraire_trades(signals: pd.DataFrame, spread: pd.Series) -> list[dict]: """ Extrait la liste des trades individuels à partir des signaux. Chaque trade contient : date d'entrée, date de sortie, type, P&L, durée. """ trades = [] trade_en_cours = None for date, row in signals.iterrows(): trade_type = row["trade_type"] if trade_type in ("entry_long", "entry_short"): # Nouvelle entrée trade_en_cours = { "date_entree": date, "type": "long" if trade_type == "entry_long" else "short", "spread_entree": float(spread.loc[date]), } elif trade_type in ("exit", "stop_loss") and trade_en_cours is not None: # Sortie du trade spread_sortie = float(spread.loc[date]) spread_entree = trade_en_cours["spread_entree"] # P&L : si long, on gagne quand le spread monte ; si short, quand il baisse if trade_en_cours["type"] == "long": pnl = spread_sortie - spread_entree else: pnl = spread_entree - spread_sortie trades.append({ "date_entree": trade_en_cours["date_entree"], "date_sortie": date, "type": trade_en_cours["type"], "sortie": trade_type, "spread_entree": round(spread_entree, 4), "spread_sortie": round(spread_sortie, 4), "pnl": round(pnl, 4), "duree_jours": (date - trade_en_cours["date_entree"]).days, }) trade_en_cours = None return trades