Antoine Boucher

Backtest d’indicateurs techniques sur plusieurs tickers avec Python

Introduction

Ce rapport présente une expérimentation sur les indicateurs techniques avec le projet BatchBacktesting sur GitHub : BatchBacktesting.

Installation des dépendances

Installez les bibliothèques nécessaires :

!pip install numpy httpx richp

Imports

Modules à importer pour le script :

import pandas as pd
import numpy as np
from datetime import datetime
import httpx
import concurrent.futures
import glob
import warnings
from rich.progress import track

warnings.filterwarnings(“ignore”)

Configuration API

Remplacez FMP_API_KEY et BINANCE_API_KEY par vos clés pour accéder aux services concernés.

BASE_URL_FMP = “https://financialmodelingprep.com/api/v3"
BASE_URL_BINANCE = “https://fapi.binance.com/fapi/v1/"
FMP_API_KEY = “YOUR_FMP_API_KEY”
BINANCE_API_KEY = “YOUR_BINANCE_API_KEY”

Fonctions de requêtes API

Ces fonctions appellent différents points de terminaison pour l’historique crypto et actions.

def make_api_request(api_endpoint, params):
with httpx.Client() as client:
response = client.get(api_endpoint, params=params)
if response.status_code == 200:
return response.json()
print(“Error: Failed to retrieve data from API”)
return None

def get_historical_price_full_crypto(symbol):
api_endpoint = f”{BASE_URL_FMP}/historical-price-full/crypto/{symbol}”
params = {“apikey”: FMP_API_KEY}
return make_api_request(api_endpoint, params)

def get_historical_price_full_stock(symbol):
api_endpoint = f"{BASE_URL_FMP}/historical-price-full/{symbol}"
params = {“apikey”: FMP_API_KEY}
return make_api_request(api_endpoint, params)

def get_SP500():
api_endpoint = “https://en.wikipedia.org/wiki/List_of_S%26P_500_companies”
data = pd.read_html(api_endpoint)
return list(data[0][‘Symbol’])

def get_all_crypto():
return [
“BTCUSD”, “ETHUSD”, “LTCUSD”, “BCHUSD”, “XRPUSD”, “EOSUSD”,
“XLMUSD”, “TRXUSD”, “ETCUSD”, “DASHUSD”, “ZECUSD”, “XTZUSD”,
“XMRUSD”, “ADAUSD”, “NEOUSD”, “XEMUSD”, “VETUSD”, “DOGEUSD”,
“OMGUSD”, “ZRXUSD”, “BATUSD”, “USDTUSD”, “LINKUSD”, “BTTUSD”,
“BNBUSD”, “ONTUSD”, “QTUMUSD”, “ALGOUSD”, “ZILUSD”, “ICXUSD”,
“KNCUSD”, “ZENUSD”, “THETAUSD”, “IOSTUSD”, “ATOMUSD”, “MKRUSD”,
“COMPUSD”, “YFIUSD”, “SUSHIUSD”, “SNXUSD”, “UMAUSD”, “BALUSD”,
“AAVEUSD”, “UNIUSD”, “RENBTCUSD”, “RENUSD”, “CRVUSD”, “SXPUSD”,
“KSMUSD”, “OXTUSD”, “DGBUSD”, “LRCUSD”, “WAVESUSD”, “NMRUSD”,
“STORJUSD”, “KAVAUSD”, “RLCUSD”, “BANDUSD”, “SCUSD”, “ENJUSD”
]

def get_financial_statements_lists():
api_endpoint = f"{BASE_URL_FMP}/financial-statement-symbol-lists"
params = {“apikey”: FMP_API_KEY}
return make_api_request(api_endpoint, params)

Stratégie EMA

La moyenne mobile exponentielle (EMA) accorde plus de poids aux points récents ; elle réagit plus vite que la moyenne mobile simple (SMA).

class EMA(Strategy):
n1 = 20
n2 = 80

def init(self):  
    close = self.data.Close  
    self.ema20 = self.I(taPanda.ema, close.s, self.n1)  
    self.ema80 = self.I(taPanda.ema, close.s, self.n2)  

def next(self):  
    price = self.data.Close  
    if crossover(self.ema20, self.ema80):  
        self.position.close()  
        self.buy(sl=0.90 \* price, tp=1.25 \* price)  
    elif crossover(self.ema80, self.ema20):  
        self.position.close()  
        self.sell(sl=1.10 \* price, tp=0.75 \* price)

Dans cette stratégie :

  • ema20 et ema80 sont calculées pour l’instrument.
  • Achat quand ema20 croise au-dessus de ema80.
  • Vente quand ema80 croise au-dessus de ema20.
  • Stop loss (sl) et take profit (tp) pour limiter les pertes et prendre des gains.

Stratégie MACD

Le MACD est un indicateur de momentum qui compare deux moyennes mobiles (souvent EMA 12 et 26) ; la ligne de signal est une EMA 9 du MACD et sert de déclencheur d’achat/vente.

class MACD(Strategy):
short_period = 12
long_period = 26
signal_period = 9

def init(self):  
    close = self.data.Close  
    self.macd = self.I(taPanda.macd, close.s, self.short\_period, self.long\_period, self.signal\_period)  

def next(self):  
    macd\_line = self.macd.macd  
    signal\_line = self.macd.signal  
    if crossover(macd\_line, signal\_line):  
        self.position.close()  
        self.buy()  
    elif crossover(signal\_line, macd\_line):  
        self.position.close()  
        self.sell()
  • macd_line et signal_line dérivent des EMA courtes et longues.
  • Achat quand macd_line croise au-dessus de signal_line.
  • Vente quand signal_line croise au-dessus de macd_line.

Exécuter les backtests

Fonctions pour traiter les instruments et lancer les stratégies choisies.

def run_backtests_strategies(instruments, strategies):
strategies = [x for x in STRATEGIES if x.__name__ in strategies]
outputs = []
with concurrent.futures.ThreadPoolExecutor() as executor:
futures = []
for strategy in strategies:
future = executor.submit(run_backtests, instruments, strategy, 4)
futures.append(future)
for future in concurrent.futures.as_completed(futures):
outputs.extend(future.result())
return outputs

def check_crypto(instrument):
return instrument in get_all_crypto()

def check_stock(instrument):
return instrument not in get_financial_statements_lists()

def process_instrument(instrument, strategy):
try:
if check_crypto(instrument):
data = get_historical_price_full_crypto(instrument)
else:
data = get_historical_price_full_stock(instrument)
if data is None or “historical” not in data:
print(f"Error processing {instrument}: No data")
return None
data = clean_data(data)
bt = Backtest(data, strategy=strategy, cash=100000, commission=0.002, exclusive_orders=True)
output = bt.run()
output = process_output(output, instrument, strategy)
return output, bt
except Exception as e:
print(f"Error processing {instrument}: {str(e)}")
return None

def clean_data(data):
data = data[“historical”]
data = pd.DataFrame(data)
data.columns = [x.title() for x in data.columns]
data = data.drop([“Adjclose”, “Unadjustedvolume”, “Change”, “Changepercent”, “Vwap”, “Label”, “Changeovertime”], axis=1)
data[“Date”] = pd.to_datetime(data[“Date”])
data.set_index(“Date”, inplace=True)
data = data.iloc[::-1]
return data

def process_output(output, instrument, strategy, in_row=True):
if in_row:
output = pd.DataFrame(output).T
output[“Instrument”] = instrument
output[“Strategy”] = strategy.__name__
output.pop("_strategy")
return output

def save_output(output, output_dir, instrument, start, end):
print(f"Saving output for {instrument}")
fileNameOutput = f"{output_dir}/{instrument}-{start}-{end}.csv"
output.to_csv(fileNameOutput)

def plot_results(bt, output_dir, instrument, start, end):
print(f"Saving chart for {instrument}")
fileNameChart = f"{output_dir}/{instrument}-{start}-{end}.html"
bt.plot(filename=fileNameChart, open_browser=False)

def run_backtests(instruments, strategy, num_threads=4, generate_plots=False):
outputs = []
output_dir = f"output/raw/{strategy.__name__}"
output_dir_charts = f"output/charts/{strategy.__name__}"
if not os.path.exists(output_dir):
os.makedirs(output_dir)
if not os.path.exists(output_dir_charts):
os.makedirs(output_dir_charts)
with concurrent.futures.ThreadPoolExecutor(max_workers=num_threads) as executor:
future_to_instrument = {executor.submit(process_instrument, instrument, strategy): instrument for instrument in instruments}
for future in concurrent.futures.as_completed(future_to_instrument):
instrument = future_to_instrument[future]
output = future.result()
if output is not None:
outputs.append(output[0])
save_output(output[0], output_dir, instrument, output[0][“Start”].to_string().strip().split()[1], output[0][“End”].to_string().strip().split()[1])
if generate_plots:
plot_results(output[1], output_dir_charts, instrument, output[0][“Start”].to_string().strip().split()[1], output[0][“End”].to_string().strip().split()[1])
data_frame = pd.concat(outputs)
start = data_frame[“Start”].to_string().strip().split()[1]
end = data_frame[“End”].to_string().strip().split()[1]
fileNameOutput = f"output/{strategy.__name__}-{start}-{end}.csv"
data_frame.to_csv(fileNameOutput)
return data_frame

Lancer les scripts

tickers = get_SP500()
run_backtests(tickers, strategy=EMA, num_threads=12, generate_plots=True)
run_backtests(tickers, strategy=MACD, num_threads=12, generate_plots=True)

ticker = get_all_crypto()
run_backtests(ticker, strategy=EMA, num_threads=12, generate_plots=True)
run_backtests(ticker, strategy=MACD, num_threads=12, generate_plots=True)

Le dossier output du dépôt BatchBacktesting ne contient en général pas de résultats précalculés — les auteurs évitent d’y versionner des données spécifiques à chaque utilisateur.

Pour obtenir des chiffres, exécutez le script localement avec vos paramètres et stratégies ; les sorties iront dans le répertoire output du projet.

Exemple de graphique de référence : EMA — AAPL.

Analyse des résultats

Exemple de classement EMA (rendements les plus hauts et plus bas) :

  • Cinq instruments avec les rendements les plus élevés :
  • BTCBUSD: 293.78%
  • ALB: 205.97%
  • OMGUSD: 199.62%
  • BBWI: 196.82%
  • GRMN: 193.47%
  • Cinq instruments avec les rendements les plus faibles :
  • BTTBUSD: -99.93%
  • UAL: -82.63%
  • NCLH: -81.51%
  • LNC: -78.02%
  • CHRW: -76.38%

Conclusion

BatchBacktesting offre une approche souple pour tester des indicateurs techniques sur actions et crypto. Les fonctions fournies s’intègrent aux APIs financières et simplifient la manipulation des données. Les résultats expérimentaux peuvent nourrir l’affinage de stratégies algorithmiques — en gardant à l’esprit sur-ajustement et réalité des frais.

Publié à l’origine sur Medium.