Introduction : Pourquoi Combiner Tardis et Backtrader ?
En tant que développeur de stratégies de trading algorithmique depuis plus de sept ans, j'ai testé des dizaines de sources de données et de frameworks de backtesting. Le组合 gagnant que j'utilise désormais pour mes projets professionnels combine Tardis.exchange pour les données tick haute fidélité et Python Backtrader pour la simulation de stratégies. Dans cet article, je vais vous montrer comment mettre en place cette architecture complète, des erreurs coûteuses que j'ai rencontrées aux solutions qui fonctionnent en production.
Les données tick sont essentielles pour tester des stratégies de scalping, d'arbitrage ou de market making qui dépendent du micro-mouvement des prix. Tardis.exchange offre un accès à plus de 35 exchanges avec des données historiques atteignant plusieurs années en arrière, avec une granularité au tick unique.
Configuration Initiale du Projet
Avant de commencer, assurezvous d'avoir un environnement Python 3.9+ avec les dépendances nécessaires. Je recommande fortement d'utiliser un environnement virtuel pour éviter les conflits de versions.
# Création de l'environnement virtuel
python -m venv backtrader-env
source backtrader-env/bin/activate # Linux/Mac
backtrader-env\Scripts\activate # Windows
Installation des dépendances
pip install backtrader[talib] pandas numpy requests websocket-client
pip install tardis-client # Client officiel Tardis
Vérification de l'installation
python -c "import backtrader; print(f'Backtrader version: {backtrader.__version__}')"
Connexion à l'API Tardis pour les Données Historiques
Tardis.exchange propose un API REST puissant pour récupérer les données historiques et un protocole WebSocket pour le temps réel. Pour le backtesting, nous utilisons principalement l'API REST qui permet de télécharger des candles ou des trades individuels sur une période définie.
Obtention de votre Clé API Tardis
Inscrivez-vous sur Tardis.exchange pour obtenir vos identifiants API. Le plan gratuit inclut 100 000 messages par mois, suffisant pour commencer vos tests. Les plans payants démarrent à 49€/mois pour 10 millions de messages.
# config.py - Configuration centralisée
import os
Clés API
TARDIS_API_KEY = os.getenv('TARDIS_API_KEY', 'your-tardis-api-key')
TARDIS_WS_URL = "wss://tardis-node-1.cluster.tardis.dev"
Paramètres de connexion
EXCHANGE = 'binance' # Exchange cible
SYMBOL = 'btcusdt' # Symbole de trading
TIMEFRAME = '1m' # Timeframe pour les bougies
START_DATE = '2024-01-01'
END_DATE = '2024-02-01'
Configuration Backtrader
INITIAL_CASH = 100000 # Capital initial en USDT
COMMISSION = 0.001 # Commission de 0.1%
# tardis_client.py - Module de connexion à Tardis
import requests
import pandas as pd
from datetime import datetime, timedelta
from typing import List, Dict, Optional
class TardisDataProvider:
"""
Provider de données historique depuis l'API Tardis.exchange
Documentation: https://docs.tardis.dev/rest-api
"""
BASE_URL = "https://tardis-api.cluster.tardis.dev/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def get_exchanges(self) -> List[Dict]:
"""Liste tous les exchanges disponibles"""
response = requests.get(
f"{self.BASE_URL}/exchanges",
headers=self.headers
)
response.raise_for_status()
return response.json()
def get_symbols(self, exchange: str) -> List[str]:
"""Récupère les symbols disponibles pour un exchange"""
response = requests.get(
f"{self.BASE_URL}/exchanges/{exchange}/symbols",
headers=self.headers
)
response.raise_for_status()
return [s['symbol'] for s in response.json()]
def download_candles(
self,
exchange: str,
symbol: str,
start_date: str,
end_date: str,
timeframe: str = '1m'
) -> pd.DataFrame:
"""
Télécharge les données de bougies OHLCV depuis Tardis
Args:
exchange: Nom de l'exchange (ex: 'binance', 'bybit')
symbol: Symbole de trading (ex: 'btcusdt')
start_date: Date de début (YYYY-MM-DD)
end_date: Date de fin (YYYY-MM-DD)
timeframe: Timeframe ('1m', '5m', '1h', '1d')
Returns:
DataFrame avec colonnes: timestamp, open, high, low, close, volume
"""
# Conversion des dates en timestamps millisecondes
start_ts = int(pd.Timestamp(start_date).timestamp() * 1000)
end_ts = int(pd.Timestamp(end_date).timestamp() * 1000)
params = {
"from": start_ts,
"to": end_ts,
"timeframe": timeframe,
"limit": 1000 # Maximum par requête
}
all_candles = []
while start_ts < end_ts:
params["from"] = start_ts
params["to"] = min(start_ts + 86400000, end_ts) # 1 jour max
response = requests.get(
f"{self.BASE_URL}/historical/candles/{exchange}:{symbol}",
headers=self.headers,
params=params
)
response.raise_for_status()
data = response.json()
if not data:
break
all_candles.extend(data)
start_ts = data[-1]['timestamp'] + 1
print(f"Téléchargé {len(data)} bougies, total: {len(all_candles)}")
# Conversion en DataFrame
df = pd.DataFrame(all_candles)
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('timestamp', inplace=True)
return df[['open', 'high', 'low', 'close', 'volume']]
def download_trades(
self,
exchange: str,
symbol: str,
start_date: str,
end_date: str
) -> pd.DataFrame:
"""
Télécharge les trades individuels (données tick)
"""
start_ts = int(pd.Timestamp(start_date).timestamp() * 1000)
end_ts = int(pd.Timestamp(end_date).timestamp() * 1000)
all_trades = []
current_ts = start_ts
while current_ts < end_ts:
params = {
"from": current_ts,
"to": min(current_ts + 3600000, end_ts), # 1 heure max
"limit": 50000
}
response = requests.get(
f"{self.BASE_URL}/historical/trades/{exchange}:{symbol}",
headers=self.headers,
params=params
)
response.raise_for_status()
data = response.json()
if not data:
break
all_trades.extend(data)
current_ts = data[-1]['timestamp'] + 1
print(f"Trades téléchargés: {len(all_trades)}")
df = pd.DataFrame(all_trades)
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
return df
Exemple d'utilisation
if __name__ == "__main__":
provider = TardisDataProvider(TARDIS_API_KEY)
# Test de connexion
exchanges = provider.get_exchanges()
print(f"Exchanges disponibles: {len(exchanges)}")
# Téléchargement d'un mois de données
df = provider.download_candles(
exchange='binance',
symbol='btcusdt',
start_date='2024-01-01',
end_date='2024-02-01',
timeframe='5m'
)
print(f"Données téléchargées: {len(df)} bougies")
print(df.head())
Intégration avec Python Backtrader
Backtrader est le framework de backtesting Python le plus populaire pour sa flexibilité et sa communauté active. Pour utiliser nos données Tardis, nous devons créer un data feed personnalisé qui étend la classe bt.feeds.PandasData.
# backtrader_datafeed.py - Data feed personnalisé pour Backtrader
import backtrader as bt
import pandas as pd
from datetime import datetime
class TardisDataFeed(bt.feeds.PandasData):
"""
Data feed Backtrader créé à partir des données Tardis
Ce data feed supporte:
- Données OHLCV standard
- Données tick avec timestamp haute résolution
- Indicateurs personnalisés
"""
# Définition des colonnes de données
params = (
('datetime', 'timestamp'),
('open', 'open'),
('high', 'high'),
('low', 'low'),
('close', 'close'),
('volume', 'volume'),
('openinterest', -1), # Non utilisé
('datetime_format', '%Y-%m-%d %H:%M:%S'),
('fromdate', None),
('todate', None),
)
class TardisTickDataFeed(bt.feeds.PandasData):
"""
Data feed pour les données tick individuelles
Plus précis pour les stratégies haute fréquence
"""
params = (
('datetime', 'timestamp'),
('datetime_format', '%Y-%m-%d %H:%M:%S.%f'),
('open', 'price'),
('high', 'price'),
('low', 'price'),
('close', 'price'),
('volume', 'amount'),
('openinterest', -1),
)
def create_data_feed(df: pd.DataFrame, timeframe=bt.TimeFrame.Minutes):
"""
Factory function pour créer un data feed Backtrader
Args:
df: DataFrame pandas avec les données OHLCV
timeframe: Timeframe Backtrader
Returns:
Instance de TardisDataFeed
"""
# Conversion de l'index en colonne si nécessaire
if not isinstance(df.index, pd.RangeIndex):
df = df.reset_index()
if 'timestamp' not in df.columns:
df.rename(columns={'index': 'timestamp'}, inplace=True)
# Création du data feed
data = TardisDataFeed()
data.df = df.copy()
return data
Création d'une Stratégie de Trading
Maintenant que nous avons notre data feed, créons une stratégie complète avec gestion du risque et journalisation détaillée. Je vais utiliser une stratégie de croisement de moyennes mobiles avec stop-loss et take-profit.
# strategy.py - Stratégie de trading avec Backtrader
import backtrader as bt
import numpy as np
from datetime import datetime
class SMACrossStrategy(bt.Strategy):
"""
Stratégie de croisement SMA avec gestion du risque
Règles:
- Achat quand SMA 10 croise au-dessus SMA 50
- Vente quand SMA 10 croise en-dessous SMA 50
- Stop-loss à 2% du prix d'entrée
- Take-profit à 5% du prix d'entrée
Paramètres:
fast_period: Période SMA rapide (default: 10)
slow_period: Période SMA lente (default: 50)
stop_loss: Pourcentage stop-loss (default: 0.02)
take_profit: Pourcentage take-profit (default: 0.05)
"""
params = (
('fast_period', 10),
('slow_period', 50),
('stop_loss', 0.02),
('take_profit', 0.05),
('printlog', True),
)
def __init__(self):
# Initialisation des indicateurs
self.sma_fast = bt.indicators.SimpleMovingAverage(
self.data.close, period=self.p.fast_period
)
self.sma_slow = bt.indicators.SimpleMovingAverage(
self.data.close, period=self.p.slow_period
)
# Croisement
self.crossover = bt.indicators.CrossOver(self.sma_fast, self.sma_slow)
# Ordres en attente
self.order = None
# Journal de trading
self.trades_log = []
def log(self, txt, dt=None):
"""Journalisation des événements"""
if self.p.printlog:
dt = dt or self.datas[0].datetime.date(0)
print(f'{dt.isoformat()} {txt}')
def notify_order(self, order):
"""Gestion des ordres"""
if order.status in [order.Submitted, order.Accepted]:
return # Ordre accepté
if order.status in [order.Completed]:
if order.isbuy():
self.log(f'ACHAT EXÉCUTÉ, Prix: {order.executed.price:.2f}, '
f'Commission: {order.executed.comm:.2f}')
self.buy_price = order.executed.price
self.buy_comm = order.executed.comm
else:
self.log(f'VENTE EXÉCUTÉE, Prix: {order.executed.price:.2f}, '
f'Commission: {order.executed.comm:.2f}')
self.order = None
elif order.status in [order.Canceled, order.Margin, order.Rejected]:
self.log('Ordre annulé/rejeté')
self.order = None
def notify_trade(self, trade):
"""Notification des trades clos"""
if not trade.isclosed:
return
pnl = trade.pnl
pnl_net = trade.pnlcomm
self.log(f'TRADE CLOS, PnL brut: {pnl:.2f}, PnL net: {pnl_net:.2f}')
# Sauvegarde dans le journal
self.trades_log.append({
'entry_date': bt.num2date(trade.dtopen),
'exit_date': bt.num2date(trade.dtclose),
'pnl': pnl,
'pnl_net': pnl_net,
'size': trade.size,
'bar_count': trade.barlen
})
def next(self):
"""Logique principale de la stratégie"""
# Vérification des ordres en attente
if self.order:
return
# Vérification de la position
if not self.position:
# Pas de position: recherche d'achat
if self.crossover > 0: # Croisement haussier
self.log(f'SIGNAL ACHAT, SMA Fast: {self.sma_fast[0]:.2f}, '
f'SMA Slow: {self.sma_slow[0]:.2f}')
# Calcul de la taille de position
target_value = self.broker.getvalue() * 0.95 # 95% du capital
size = int(target_value / self.data.close[0])
self.order = self.buy(size=size)
else:
# Position existante: recherche de vente
current_price = self.data.close[0]
buy_price = self.buy_price
# Stop-loss
stop_price = buy_price * (1 - self.p.stop_loss)
if current_price <= stop_price:
self.log(f'STOP-LOSS触发, Prix: {current_price:.2f}')
self.order = self.close()
return
# Take-profit
target_price = buy_price * (1 + self.p.take_profit)
if current_price >= target_price:
self.log(f'TAKE-PROFIT触发, Prix: {current_price:.2f}')
self.order = self.close()
return
# Croisement baissier
if self.crossover < 0:
self.log(f'SIGNAL VENTE, Prix: {current_price:.2f}')
self.order = self.close()
def stop(self):
"""Fin du backtest"""
self.log(f'Capital final: {self.broker.getvalue():.2f}', dt=datetime.now())
def run_backtest(
df: pd.DataFrame,
initial_cash: float = 100000,
commission: float = 0.001,
**strategy_params
) -> dict:
"""
Exécute le backtest avec les paramètres spécifiés
Returns:
Dictionary avec résultats et statistiques
"""
# Création du cerebro
cerebro = bt.Cerebro()
# Ajout du data feed
data_feed = TardisDataFeed(
dataname=df,
datetime=None,
open='open',
high='high',
low='low',
close='close',
volume='volume',
openinterest=-1
)
cerebro.adddata(data_feed)
# Configuration du broker
cerebro.broker.setcash(initial_cash)
cerebro.broker.setcommission(commission=commission)
# Ajout de la stratégie
cerebro.addstrategy(SMACrossStrategy, **strategy_params)
# Configuration de l'analyse
cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')
cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown')
cerebro.addanalyzer(bt.analyzers.Returns, _name='returns')
cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name='trades')
cerebro.addanalyzer(bt.analyzers.TimeReturn, _name='time_return')
# Exécution
print(f'Capital initial: {initial_cash:.2f}')
results = cerebro.run()
strategy = results[0]
# Récupération des résultats
final_value = cerebro.broker.getvalue()
sharpe = strategy.analyzers.sharpe.get_analysis().get('sharperatio', None)
drawdown = strategy.analyzers.drawdown.get_analysis()
returns = strategy.analyzers.returns.get_analysis()
trades = strategy.analyzers.trades.get_analysis()
# Calcul des métriques
total_return = (final_value - initial_cash) / initial_cash * 100
max_dd = drawdown.get('max', {}).get('drawdown', 0)
# Affichage des résultats
print('\n' + '='*60)
print('RÉSULTATS DU BACKTEST')
print('='*60)
print(f'Capital final: {final_value:.2f}')
print(f'Rendement total: {total_return:.2f}%')
print(f'Sharpe Ratio: {sharpe:.3f}' if sharpe else 'Sharpe Ratio: N/A')
print(f'Drawdown max: {max_dd:.2f}%')
print(f'Trades totaux: {trades.get("total", {}).get("total", 0)}')
print('='*60)
return {
'final_value': final_value,
'total_return': total_return,
'sharpe': sharpe,
'max_drawdown': max_dd,
'trades': trades,
'strategy': strategy
}
Script Principal de Backtesting
# main.py - Script principal de backtesting
import os
import pandas as pd
from datetime import datetime
from tardis_client import TardisDataProvider
from strategy import run_backtest, SMACrossStrategy
Import des configurations
from config import (
TARDIS_API_KEY, EXCHANGE, SYMBOL,
TIMEFRAME, START_DATE, END_DATE,
INITIAL_CASH, COMMISSION
)
def main():
"""
Script principal: téléchargement des données et exécution du backtest
"""
print("="*60)
print("BACKTESTING TARDIS + BACKTRADER")
print("="*60)
# Initialisation du provider
provider = TardisDataProvider(TARDIS_API_KEY)
# Téléchargement des données
print(f"\nTéléchargement des données...")
print(f"Exchange: {EXCHANGE}")
print(f"Symbole: {SYMBOL}")
print(f"Période: {START_DATE} -> {END_DATE}")
print(f"Timeframe: {TIMEFRAME}")
try:
df = provider.download_candles(
exchange=EXCHANGE,
symbol=SYMBOL,
start_date=START_DATE,
end_date=END_DATE,
timeframe=TIMEFRAME
)
print(f"\nDonnées téléchargées: {len(df)} bougies")
print(f"Période réelle: {df.index.min()} -> {df.index.max()}")
# Affichage des statistiques descriptives
print("\nStatistiques des prix:")
print(df.describe())
# Sauvegarde des données
cache_file = f"data_{EXCHANGE}_{SYMBOL}_{START_DATE}_{END_DATE}.csv"
df.to_csv(cache_file)
print(f"\nDonnées sauvegardées dans: {cache_file}")
except Exception as e:
print(f"Erreur lors du téléchargement: {e}")
# Tentative de chargement depuis le cache
cache_file = f"data_{EXCHANGE}_{SYMBOL}_{START_DATE}_{END_DATE}.csv"
if os.path.exists(cache_file):
print(f"Chargement depuis le cache: {cache_file}")
df = pd.read_csv(cache_file, index_col=0, parse_dates=True)
else:
raise
# Exécution du backtest
print("\n" + "="*60)
print("EXÉCUTION DU BACKTEST")
print("="*60)
results = run_backtest(
df=df,
initial_cash=INITIAL_CASH,
commission=COMMISSION,
fast_period=10,
slow_period=50,
stop_loss=0.02,
take_profit=0.05,
printlog=True
)
# Optimisation des paramètres
print("\n" + "="*60)
print("OPTIMISATION DES PARAMÈTRES")
print("="*60)
fast_periods = [5, 10, 15, 20]
slow_periods = [30, 50, 100]
best_params = None
best_return = float('-inf')
for fast in fast_periods:
for slow in slow_periods:
if fast >= slow:
continue
print(f"\nTest: SMA({fast}, {slow})")
result = run_backtest(
df=df,
initial_cash=INITIAL_CASH,
commission=COMMISSION,
fast_period=fast,
slow_period=slow,
stop_loss=0.02,
take_profit=0.05,
printlog=False
)
if result['total_return'] > best_return:
best_return = result['total_return']
best_params = (fast, slow)
print(f" -> NOUVEAU MEILLEUR: {best_return:.2f}%")
print("\n" + "="*60)
print("MEILLEURS PARAMÈTRES")
print("="*60)
print(f"Périodes SMA: {best_params}")
print(f"Rendement: {best_return:.2f}%")
if __name__ == "__main__":
main()
Comparatif : Tardis vs Alternatives pour le Backtesting
| Caractéristique | Tardis.exchange | CCXT + Exchange API | Binance Historical |
|---|---|---|---|
| Multi-exchange | 35+ exchanges | Nombreux exchanges | Binance uniquement |
| Historique données | 2-3 années | Variable (limité) | 1-2 années |
| Granularité tick | ✓ Tick complet | Variable | Candles uniquement |
| Fiabilité | Haute (infrastructure dédiée) | Moyenne | Haute |
| API unifiée | ✓ Oui | ✓ Oui | ✗ Non |
| Plan gratuit | 100K messages/mois | 0 (rate limits) | Illimité |
| Prix adulte | 49€/mois | Gratuit | Gratuit |
Erreurs Courantes et Solutions
Erreur 1 : "403 Forbidden" lors de l'accès à l'API Tardis
Symptôme : La requête API retourne un code d'erreur 403 avec le message "Forbidden" ou "Invalid API key".
Causes possibles :
- Clé API incorrecte ou mal formatée
- Expiration du plan gratuit (quotas dépassés)
- Restrictions géographiques
Solution :
# Vérification et correction de la clé API
import os
Méthode 1: Variable d'environnement
TARDIS_API_KEY = os.getenv('TARDIS_API_KEY')
if not TARDIS_API_KEY:
print("ERREUR: TARDIS_API_KEY non définie")
exit(1)
Méthode 2: Vérification du format
if not TARDIS_API_KEY.startswith(('tk_live_', 'tk_test_')):
print("ATTENTION: Format de clé API inhabituel")
print(f"Clé actuelle: {TARDIS_API_KEY[:10]}...")
Méthode 3: Test de connexion
import requests
response = requests.get(
"https://tardis-api.cluster.tardis.dev/v1/exchanges",
headers={"Authorization": f"Bearer {TARDIS_API_KEY}"}
)
if response.status_code == 200:
print("Connexion API réussie ✓")
elif response.status_code == 403:
print("ERREUR 403: Vérifiez votre clé API ou quotas")
print("Consultez: https://tardis.dev/account")
else:
print(f"Erreur: {response.status_code} - {response.text}")
Erreur 2 : "IndexError: index out of range" dans Backtrader
Symptôme : L'erreur se produit lors du calcul des indicateurs ou de l'accès aux données.
Cause : Les données sont insuffisantes pour calculer les indicateurs (ex: SMA 50 sur moins de 50 bougies).
Solution :
# Validation des données avant backtest
import pandas as pd
def validate_data_for_backtest(df: pd.DataFrame, min_periods: int = 100) -> bool:
"""
Valide que les données sont suffisantes pour le backtest
Args:
df: DataFrame avec les données OHLCV
min_periods: Nombre minimum de bougies requises
Returns:
True si les données sont valides
"""
# Vérification de la longueur
if len(df) < min_periods:
print(f"ERREUR: Données insuffisantes ({len(df)} < {min_periods})")
return False
# Vérification des valeurs nulles
null_count = df.isnull().sum().sum()
if null_count > 0:
print(f"ATTENTION: {null_count} valeurs nulles détectées")
print("Remplacement par forward fill...")
df.fillna(method='ffill', inplace=True)
df.fillna(method='bfill', inplace=True)
# Vérification de la continuité temporelle
if isinstance(df.index, pd.DatetimeIndex):
time_diff = df.index.to_series().diff().median()
print(f"Intervalle temporel médian: {time_diff}")
if pd.isna(time_diff) or time_diff == pd.Timedelta(0):
print("ERREUR: Données non continues")
return False
# Vérification des prix
if (df['close'] <= 0).any():
print("ERREUR: Prix négatifs ou nuls détectés")
return False
print(f"Validation réussie: {len(df)} bougies")
print(f"Période: {df.index.min()} -> {df.index.max()}")
print(df.describe())
return True
Utilisation dans le script principal
df = provider.download_candles(...)
if validate_data_for_backtest(df, min_periods=100):
results = run_backtest(df)
else:
print("Impossible de lancer le backtest")
Erreur 3 : "Commission non appliquée correctement"
Symptôme : Le capital final ne correspond pas aux attentes après application des commissions.
Cause : Configuration incorrecte de la commission ou confusion entre commission fixe et pourcentage.
Solution :
# Configuration correcte des commissions
import backtrader as bt
Méthode 1: Commission pourcentage (0.1% = 0.001)
cerebro.broker.setcommission(commission=0.001) # Par trade
Méthode 2: Commission fixe + pourcentage
cerebro.broker.setcommission(
commission=0.0002, # 0.02% du volume
mult=1.0, # Multiplicateur
margin=1000, # Marge requise (pour futures)
commtype=bt.COMM_PERC # Type: COMM_PERC ou COMM_FIXED
)
Méthode 3: Classe de commission personnalisée
class CryptoCommission(bt.CommInfo):
"""Commission personnalisée pour crypto avec slippage"""
params = (
('commission', 0.001), # 0.1%
('slippage', 0.0005), # 0.05% slippage
('min_commission', 0.5), # Commission minimum USDT
)
def _getcommission(self, size, price, pseudoexec=False):
"""Calcul de la commission"""
commission = abs(size) * price * self.params.commission
return max(commission, self.params.min_commission)
def getslippage(self, size, price):
"""Calcul du slippage"""
return price * self.params.slippage * abs(size)
Application
comminfo = CryptoCommission()
cerebro.broker.addcommissioninfo(comminfo)
Vérification
print(f"Commission configurée: {cerebro.broker.getcommissioninfo}")
print(f"Vérification: 1 BTC à 50000 avec 0.1% = {50000 * 0.001} USDT")
Optimisation Avanzada del Backtest
Pour les traders sérieux, voici quelques optimisations que j'utilise en production pour améliorer la précision du backtesting :
Walk-Forward Analysis
Cette technique divise les données en périodes d'entraînement et de test pour éviter le sur-apprentissage.
# walk_forward.py - Analyse walk-forward
import numpy as np
import pandas as pd
from itertools import product
def walk_forward_optimization(
df: pd.DataFrame,
param_grid: dict,
train_size: int = 500,
test_size: int = 100,
step: int = 50
) -> pd.DataFrame:
"""
Optimisation walk-forward pour éviter le sur-apprentissage
Args:
df: DataFrame avec les données complètes
param_grid: Grille de paramètres à tester
train_size: Taille de la période d'entraînement
test_size: Taille de la période de test
step: Pas de déplacement de la fenêtre
Returns:
DataFrame avec les résultats de chaque fenêtre
"""
results = []
total_windows = (len(df) - train_size - test_size) // step + 1
for i in range(total_windows):
train_start = i * step
train_end = train_start + train_size
test_start = train_end
test_end = min(test_start + test_size, len(df))
train_df = df.iloc[train_start:train_end]
test_df = df.iloc[test_start:test_end]
print(f"\nFenêtre {i+1}/{total_windows}")
print(f" Entraînement: {train_df.index[0]} -> {train_df.index[-1]}")
print(f" Test: {test_df.index[0]} -> {test_df.index[-1]}")
# Grid search sur les données d'entraînement
best_params = None
best_train_return = float('-inf')
param_combinations = list(product(
*[param_grid[k] for k in sorted(param_grid.keys())]
))
param_names = sorted(param_grid.keys())
for params in param_combinations:
param_dict = dict(zip(param_names, params))
try:
result = run_backtest(
train_df,
initial_cash=INITIAL_CASH,
commission=COMMISSION,
printlog=False,
**param_dict
)
if result['total_return'] > best_train_return:
best_train_return = result['total_return']
best_params = param_dict
except Exception as e:
continue
# Test avec les meilleurs paramètres
test_result = run_backtest(
test_df,
initial_cash=INITIAL_CASH,
commission=COMMISSION,
printlog=False,
**best_params
)
results.append({
'window': i + 1,
'train_period': f"{train_df.index[0].date()} - {train_df.index[-1].date()}",
'test_period': f"{test_df.index[0].date()} - {test_df.index[-1].date()}",
'best_params': best_params,
'train_return': best_train_return,
'test_return': test_result['total_return'],
'test_sharpe': test_result['sharpe']
})
print(f" Meilleurs params: {best_params}")
print(f" Train return: {best_train_return:.2f}%")
print