Optimisation de portefeuille sur le Forex : Intégration de la VaR et de la théorie de Markowitz

Introduction : Optimisation de portefeuille sur le Forex

J'ai passé les trois dernières années à développer des robots de trading pour le Forex. Et vous savez quoi ? La gestion des risques, c'est vraiment pénible. Au début, je me contentais de placer des stops jusqu'à ce que je perde quelques dépôts. Puis j’ai commencé à creuser davantage le sujet et je suis tombé sur la théorie de l'optimisation de portefeuille de Markowitz.

Ça avait l'air bien : on calcule les corrélations, on optimise les pondérations... Mais en réalité, cela ne fonctionne pas très bien pour le Forex. Mais pourquoi ? Car en Forex, toutes les paires sont interdépendantes ! Essayez de trader simultanément l'EUR/USD et l'EUR/GBP et vous comprendrez ce que je veux dire. Un mouvement brusque de l'EUR et les deux positions se mettent à évoluer presque comme une seule position. Une belle théorie se brise face à une dure réalité.

N'en pouvant plus, j'ai commencé à chercher d'autres solutions. Finalement, je suis tombé sur la méthodologie de la Valeur au Risque (Value at Risk, VaR). Au début, je ne comprenais même pas de quoi il s'agissait – un tas d’équations compliquées. Mais soudain, j'ai réalisé : c'est exactement ce dont j'avais besoin ! La VaR indique la perte maximale pour une probabilité donnée. Autrement dit, nous pouvons estimer directement combien d'argent nous pourrions perdre en une journée/semaine/mois.

Finalement, j'ai décidé de croiser Markowitz avec la VaR. Ça paraît fou ? Peut-être. Mais je ne voyais pas d'autres options. Markowitz assure une allocation optimale des fonds, et la VaR réduit le risque d’appel de marge. Sur le papier, ça paraissait parfait.

Commença alors le quotidien difficile d'un développeur‑chercheur. Python, terminal MetaTrader 5, des tonnes de données historiques... Je savais que ce ne serait pas facile, mais la réalité a dépassé toutes mes attentes. Voilà ce que je vais vous expliquer : comment j'ai essayé de créer un système qui fonctionne réellement, et qui ne produit pas seulement de beaux résultats sur les backtests.

Si vous avez déjà essayé d'automatiser le trading du Forex, vous comprendrez ma souffrance. Et sinon, peut-être que mon expérience vous aidera à éviter au moins certains des pièges sur lesquels on finit inévitablement par marcher.

Fondements théoriques et mathématiques de la VaR et de la théorie de Markowitz

Commençons donc par la théorie. Le premier mois, j'essayais simplement de me familiariser avec les mathématiques. La théorie de Markowitz semble compliquée : un tas d'équations, de matrices, d'optimisation quadratique... Mais en réalité, tout est simple : on prend les rendements des actifs, on calcule les corrélations et on trouve les pondérations de sorte que le risque soit minimal pour un rendement donné.

Au début, j'étais content ! Mais ensuite j'ai commencé à faire des tests avec de vraies données du Forex, et là tout a commencé... En utilisant l'historique de l'EUR/USD sur un an, la distribution des rendements n'était absolument pas normale. Il en était de même pour la paire GBP/USD. Il s'agit là de l'hypothèse fondamentale de la théorie de Markowitz. Autrement dit, tous ces calculs tombent à l’eau.

J'ai passé une semaine à chercher une solution. J'ai épluché des articles scientifiques, fait des recherches sur Google, lu des forums. Je suis retourné à mon article sur la VaR (Value at Risk). Cela paraît intelligent, mais en réalité, nous calculons simplement combien nous pouvons perdre avec une probabilité de 95% (ou autre). J'ai d'abord essayé l'option la plus simple : la VaR paramétrique. L'équation est élémentaire : moyenne moins sigma par quantile. Mais la qualité des résultats est médiocre.

Je suis ensuite passé à la VaR historique. L'idée est de partir de données historiques réelles et d'examiner quelles ont été les pertes dans les 5% des cas les plus défavorables. C'est beaucoup plus proche de la réalité, mais il faut beaucoup de données. Le boss final, c'est la méthode de Monte Carlo. Nous générons un tas de scénarios aléatoires en tenant compte des corrélations entre les paires, et finalement nous obtenons quelque chose de sensé.

Le plus difficile a été de trouver comment combiner la VaR avec l'optimisation de Markowitz. De ce fait, voici ce qui est né : nous prenons l'optimisation standard, mais nous y ajoutons une contrainte portant sur la VaR. Nous recherchons le risque minimal pour un rendement donné, tout en veillant à ce que la VaR ne dépasse pas un certain seuil.

Sur le papier, tout est parfait, mais il faut le programmer... Dans les sections suivantes, je montrerai comment j'ai transformé ces équations en un code Python fonctionnel.

Connexion à MetaTrader 5 depuis Python

L’implémentation de mon système a commencé par l'établissement d'une connexion stable avec le terminal de trading. Après avoir expérimenté différentes approches, j'ai opté pour une connexion directe via la bibliothèque MetaTrader 5 pour Python, qui s'est avérée être la plus fiable et la plus rapide.

import MetaTrader5 as mt5
import time

def initialize_mt5(account=12345, server="MetaQuotes-Demo", password="abc123"):
    if not mt5.initialize():
        print(f"initialize() failed, error code = {mt5.last_error()}")
        return False
        
    authorized = mt5.login(account, password=password, server=server)
    if not authorized:
        print(f"login failed, error code = {mt5.last_error()}")
        mt5.shutdown()
        return False
        
    return True

Un autre problème complexe consistait à synchroniser l'heure entre le serveur du courtier et le système local. Une différence de quelques secondes pourrait entraîner de sérieux problèmes lors du calcul de la VaR. Il a été nécessaire de mettre en œuvre un mécanisme de correction spécial :

def get_time_correction():
    server_time = mt5.symbol_info_tick("EURUSD").time
    local_time = int(time.time())
    return server_time - local_time

def get_corrected_time():
    correction = get_time_correction()
    return int(time.time()) + correction

J’ai consacré beaucoup de temps à optimiser l'acquisition des données. Au départ, j'effectuais des requêtes pour chaque paire de devises séparément. Mais après la mise en place du traitement par lots, la vitesse a été multipliée plusieurs fois :

def fetch_data_batch(symbols, timeframe, start_pos, count):
    data = {}
    for symbol in symbols:
        rates = mt5.copy_rates_from_pos(symbol, timeframe, start_pos, count)
        if rates is not None and len(rates) > 0:
            data[symbol] = rates
        else:
            print(f"Failed to get data for {symbol}")
            return None
    return data

Il s'est avéré étonnamment difficile de terminer correctement le programme. Il a été nécessaire de mettre en place une procédure d’arrêt propre et sécurisé (« graceful shutdown ») :

def safe_shutdown():
    try:
        positions = mt5.positions_get()
        if positions:
            for position in positions:
                close_position(position.ticket)
        
        orders = mt5.orders_get()
        if orders:
            for order in orders:
                mt5.order_send(request={"action": mt5.TRADE_ACTION_REMOVE,
                                      "order": order.ticket})
    finally:
        mt5.shutdown()

Le résultat fut une base fiable pour l'ensemble du système, capable de fonctionner 24 heures sur 24 sans panne. Il était déjà possible de construire une logique d'optimisation de portefeuille plus complexe sur cette base. Mais c'est déjà le sujet de la section suivante.

Obtention des données historiques et leur prétraitement

Au fil des années passées à travailler avec des données de marché, j'ai appris une vérité simple : la qualité des données historiques est essentielle à tout système de trading. Surtout en matière d'optimisation de portefeuille, où les erreurs de données peuvent avoir des effets en cascade.

J'ai commencé par créer un système de chargement de l'historique fiable. La première version était assez simple, mais la pratique a rapidement révélé ses lacunes. Les cotations peuvent contenir des gaps de cotations, des pics, et parfois même des valeurs tout à fait incorrectes. Voici à quoi ressemble le code final pour le téléchargement avec une validation de base :

def load_historical_data(symbols, timeframe, start_date, end_date):
    data_frames = {}
    for symbol in symbols:
        # Load with a reserve to compensate for gaps
        rates = mt5.copy_rates_range(symbol, timeframe, 
                                   start_date - timedelta(days=30), 
                                   end_date)
        if rates is None:
            print(f"Failed to load data for {symbol}")
            continue
            
        df = pd.DataFrame(rates)
        df['time'] = pd.to_datetime(df['time'], unit='s')
        df.set_index('time', inplace=True)
        
        # Basic anomaly check
        df = detect_and_remove_spikes(df)
        df = fill_gaps(df)
        data_frames[symbol] = df
    
    return data_frames

Un autre problème résidait dans la gestion des gaps de week-end. Au départ, j'ai simplement supprimé ces jours, mais cela a faussé les calculs de volatilité. Après de longues expérimentations, une méthode d'interpolation a vu le jour, prenant en compte les spécificités de chaque paire de devises :

def fill_gaps(df, method='time'):
    if df.empty:
        return df
        
    # Check the intervals between points
    time_delta = df.index.to_series().diff()
    gaps = time_delta[time_delta > pd.Timedelta(hours=2)].index
    
    for gap_start in gaps:
        gap_end = df.index[df.index.get_loc(gap_start) + 1]
        # Create new points with interpolated values
        new_points = pd.date_range(gap_start, gap_end, freq='1H')[1:-1]
        
        for point in new_points:
            df.loc[point] = df.asof(point)
    
    return df.sort_index()

J'ai essayé plusieurs méthodes pour calculer les rendements. Les simples variations de pourcentage se sont avérées trop sensibles au bruit. Les rendements logarithmiques ont donné les meilleurs résultats pour l'estimation de la VaR :

def calculate_returns(df):
    df['returns'] = np.log(df['close'] / df['close'].shift(1))
    df['rolling_std'] = df['returns'].rolling(window=20).std()
    df['rolling_mean'] = df['returns'].rolling(window=20).mean()
    
    # Clean out emissions using the 3-sigma rule
    mean = df['returns'].mean()
    std = df['returns'].std()
    df = df[abs(df['returns'] - mean) <= 3 * std]
    
    return df

Le développement du système de vérification des données s'est avéré être une étape importante. Chaque ensemble est soumis à un contrôle en plusieurs étapes avant d'être utilisé dans les calculs :

def verify_data_quality(df, symbol):
    checks = {
        'missing_values': df.isnull().sum().sum() == 0,
        'price_continuity': (df['close'] > 0).all(),
        'timestamp_uniqueness': df.index.is_unique,
        'reasonable_returns': abs(df['returns']).max() < 0.1
    }
    
    if not all(checks.values()):
        failed_checks = [k for k, v in checks.items() if not v]
        print(f"Data quality issues for {symbol}: {failed_checks}")
        return False
        
    return True

J'ai accordé une attention particulière à la gestion des anomalies de marché. Divers événements, tels que des mouvements brusques dus à l'actualité ou des krachs éclair, peuvent fortement fausser l'évaluation des risques. J'ai développé un algorithme spécial pour les identifier et les traiter correctement :

def detect_market_anomalies(df, window=20, threshold=3):
    volatility = df['returns'].rolling(window=window).std()
    typical_range = volatility.mean() + threshold * volatility.std()
    
    anomalies = df[abs(df['returns']) > typical_range].index
    if len(anomalies) > 0:
        print(f"Detected {len(anomalies)} market anomalies")
        
    return anomalies

Le résultat fut un pipeline de traitement des données fiable qui devint la base de tous les calculs ultérieurs. Des données historiques de qualité constituent le fondement sans lequel il est impossible de construire un système de gestion de portefeuille efficace. Dans la section suivante, j'examinerai comment ces données sont utilisées pour calculer la VaR.

Implémentation du calcul de la VaR pour les paires de devises

Après avoir longtemps travaillé avec des données historiques, je me suis plongé dans la mise en œuvre du calcul de la VaR. Au départ, il semblait suffisant de prendre des équations toutes faites et de les traduire en code. La réalité s'est avérée plus complexe, car les spécificités du Forex ont nécessité de sérieuses modifications des approches standard.

J'ai commencé par implémenter trois méthodes classiques de calcul de la VaR. Voici à quoi ressemble l'approche paramétrique :

def parametric_var(returns, confidence_level=0.95, holding_period=1):
    mu = returns.mean()
    sigma = returns.std()
    z_score = norm.ppf(1 - confidence_level)
    
    daily_var = -(mu + z_score * sigma)
    return daily_var * np.sqrt(holding_period)

Mais il est rapidement apparu que l'hypothèse d'une distribution normale des rendements sur le marché des changes ne se vérifie pas souvent. L'approche historique s'est avérée plus fiable :

def historical_var(returns, confidence_level=0.95, holding_period=1):
    sorted_returns = np.sort(returns)
    index = int((1 - confidence_level) * len(sorted_returns))
    daily_var = -sorted_returns[index]
    return daily_var * np.sqrt(holding_period)

Mais les résultats les plus intéressants ont été obtenus par la méthode de Monte Carlo. Je l'ai modifiée pour tenir compte des spécificités du marché des changes :

def monte_carlo_var(returns, confidence_level=0.95, holding_period=1, simulations=10000):
    mu = returns.mean()
    sigma = returns.std()
    
    # Consider auto correlation of returns
    corr = returns.autocorr()
    
    simulated_returns = []
    for _ in range(simulations):
        daily_returns = []
        last_return = returns.iloc[-1]
        
        for _ in range(holding_period):
            # Generate the next value taking auto correlation into account
            innovation = np.random.normal(0, 1)
            next_return = mu + corr * (last_return - mu) + sigma * np.sqrt(1 - corr**2) * innovation
            daily_returns.append(next_return)
            last_return = next_return
            
        total_return = sum(daily_returns)
        simulated_returns.append(total_return)
    
    return -np.percentile(simulated_returns, (1 - confidence_level) * 100)

J'ai accordé une attention particulière à la validation des résultats. J'ai également développé un système de backtesting pour vérifier la précision de la VaR :

def backtest_var(returns, var, confidence_level=0.95):
    violations = (returns < -var).sum()
    expected_violations = len(returns) * (1 - confidence_level)
    
    z_score = (violations - expected_violations) / np.sqrt(expected_violations)
    p_value = 1 - norm.cdf(abs(z_score))
    
    return {
        'violations': violations,
        'expected': expected_violations,
        'z_score': z_score,
        'p_value': p_value
    }

Pour tenir compte des corrélations entre les paires de devises, il a été nécessaire de mettre en œuvre le calcul de la VaR du portefeuille :

def portfolio_var(returns_df, weights, confidence_level=0.95, method='historical'):
    if method == 'parametric':
        portfolio_returns = returns_df.dot(weights)
        return parametric_var(portfolio_returns, confidence_level)
    
    elif method == 'historical':
        portfolio_returns = returns_df.dot(weights)
        return historical_var(portfolio_returns, confidence_level)
    
    elif method == 'monte_carlo':
        # Use the covariance matrix to generate
        # correlated random variables
        cov_matrix = returns_df.cov()
        L = np.linalg.cholesky(cov_matrix)
        
        means = returns_df.mean().values
        simulated_returns = []
        
        for _ in range(10000):
            Z = np.random.standard_normal(len(weights))
            R = means + L @ Z
            portfolio_return = weights @ R
            simulated_returns.append(portfolio_return)
            
        return -np.percentile(simulated_returns, (1 - confidence_level) * 100)

Le résultat fut un système de calcul de la VaR flexible, adapté aux spécificités du Forex. Dans la section suivante, j'expliquerai comment ces calculs s'intègrent à la théorie de Markowitz pour l'optimisation de portefeuille.

Optimisation de portefeuille par la méthode de Markowitz

Après avoir mis en place un calcul fiable de la VaR, j'ai commencé à me concentrer sur l'optimisation du portefeuille. La théorie classique de Markowitz nécessitait une adaptation sérieuse aux réalités du Forex. Des mois d'expérimentation et de tests m'ont permis de faire plusieurs découvertes importantes.

La première chose que j'ai constatée, c'est que les indicateurs de risque et de rendement standard fonctionnent différemment sur le Forex et sur le marché boursier. Les paires de devises entretiennent des relations complexes qui évoluent au fil du temps. Après de nombreuses expérimentations, j'ai mis au point une fonction modifiée de calcul du rendement attendu :

def calculate_expected_returns(returns_df, method='ewma', halflife=30):
    if method == 'ewma':
        # Exponentially weighted average gives more weight to recent data
        return returns_df.ewm(halflife=halflife).mean().iloc[-1]
    elif method == 'capm':
        # Modified CAPM for Forex
        risk_free_rate = 0.02  # annual risk-free rate
        market_returns = returns_df.mean(axis=1)  # market returns proxy
        betas = calculate_currency_betas(returns_df, market_returns)
        return risk_free_rate + betas * (market_returns.mean() - risk_free_rate)

Le calcul de la matrice de covariance a également nécessité quelques modifications. L'approche historique simpliste a donné des résultats trop instables. J’ai implémenté une estimation de covariance par rétrécissement (shrinkage), ce qui a considérablement amélioré la robustesse de l'optimisation :

def shrinkage_covariance(returns_df, shrinkage_factor=None):
    sample_cov = returns_df.cov()
    n_assets = len(returns_df.columns)
    
    # The target matrix is diagonal with average variance
    target = np.diag(np.repeat(sample_cov.values.trace() / n_assets, n_assets))
    
    if shrinkage_factor is None:
        # Estimation of the optimal 'shrinkage' ratio
        shrinkage_factor = estimate_optimal_shrinkage(returns_df, sample_cov, target)
    
    shrunk_cov = (1 - shrinkage_factor) * sample_cov + shrinkage_factor * target
    return pd.DataFrame(shrunk_cov, index=sample_cov.index, columns=sample_cov.columns)

Le plus difficile a été d’optimiser les poids du portefeuille. Après de nombreux essais, j'ai opté pour un algorithme de programmation quadratique modifié :

def optimize_portfolio(returns_df, expected_returns, covariance, target_return=None, constraints=None):
    n_assets = len(returns_df.columns)
    
    # Risk minimization function
    def portfolio_volatility(weights):
        return np.sqrt(weights.T @ covariance @ weights)
    
    # Limitations
    constraints = []
    # The sum of the weights is 1
    constraints.append({'type': 'eq', 'fun': lambda x: np.sum(x) - 1})
    
    if target_return is not None:
        # Target income limit
        constraints.append({
            'type': 'eq',
            'fun': lambda x: x @ expected_returns - target_return
        })
    
    # Add leverage restrictions for Forex
    constraints.append({
        'type': 'ineq',
        'fun': lambda x: 20 - np.sum(np.abs(x))  # max leverage 20
    })
    
    # Initial approximation - equal weights
    initial_weights = np.repeat(1/n_assets, n_assets)
    
    # Optimization
    result = minimize(
        portfolio_volatility,
        initial_weights,
        method='SLSQP',
        constraints=constraints,
        bounds=tuple((0, 1) for _ in range(n_assets))
    )
    
    if not result.success:
        raise OptimizationError("Failed to optimize portfolio: " + result.message)
        
    return result.x

J'ai accordé une attention particulière au problème de la stabilité de la solution. De petites modifications des données d'entrée ne devraient pas entraîner une révision radicale du portefeuille. À cette fin, j'ai développé la procédure de régularisation :

def regularized_optimization(returns_df, current_weights, lambda_reg=0.1):
    # Add a penalty for deviation from the current weights
    def objective(weights):
        volatility = portfolio_volatility(weights)
        turnover_penalty = lambda_reg * np.sum(np.abs(weights - current_weights))
        return volatility + turnover_penalty

Nous disposons donc maintenant d'un optimiseur de portefeuille fiable qui tient compte des spécificités du Forex et ne nécessite pas de rééquilibrage fréquent. Mais le plus important restait à venir : combiner cette approche avec un système de contrôle des risques basé sur la VaR.

Combiner la VaR et Markowitz en un seul modèle

Combiner les deux approches s'est avéré être la partie la plus difficile. Il me fallait trouver un moyen d'utiliser les avantages des deux méthodes sans créer de contradictions entre elles. Après plusieurs mois d'expérimentation, j'ai trouvé une solution élégante.

L'idée principale était d'utiliser la VaR comme contrainte supplémentaire dans le problème d'optimisation de Markowitz. Voici ce que cela donne dans le code :

def integrated_portfolio_optimization(returns_df, target_return, max_var_limit, current_weights=None):
    n_assets = len(returns_df.columns)
    
    # Calculation of basic metrics
    exp_returns = calculate_expected_returns(returns_df)
    covariance = shrinkage_covariance(returns_df)
    
    def objective_function(weights):
        # Portfolio standard deviation (Markowitz)
        portfolio_std = np.sqrt(weights.T @ covariance @ weights)
        
        # component VaR
        portfolio_var = calculate_portfolio_var(returns_df, weights)
        var_penalty = max(0, portfolio_var - max_var_limit)
        
        return portfolio_std + 100 * var_penalty  # Penalty for exceeding VaR

Pour tenir compte de la nature dynamique du marché, j'ai développé un système adaptatif de recalcul des paramètres :

def adaptive_risk_limits(returns_df, base_var_limit, window=60):
    # Adapting VaR limits to current volatility
    recent_vol = returns_df.tail(window).std()
    long_term_vol = returns_df.std()
    vol_ratio = recent_vol / long_term_vol
    
    adjusted_var_limit = base_var_limit * np.sqrt(vol_ratio)
    return min(adjusted_var_limit, base_var_limit * 1.5)  # Limit growth

Une attention particulière a dû être portée au problème de la stabilité de la solution. J'ai mis en place un mécanisme permettant une transition en douceur entre les différents états du portefeuille :

def smooth_rebalancing(old_weights, new_weights, max_change=0.1):
    weight_diff = new_weights - old_weights
    excess_change = np.abs(weight_diff) - max_change
    
    where_excess = excess_change > 0
    if where_excess.any():
        # Limit changes in weights
        adjustment = np.sign(weight_diff) * np.minimum(
            np.abs(weight_diff),
            np.where(where_excess, max_change, np.abs(weight_diff))
        )
        return old_weights + adjustment
    return new_weights

J'ai mis au point une mesure spécifique pour évaluer l'efficacité de l'approche combinée :

def evaluate_integrated_model(returns_df, weights, var_limit):
    # Calculation of performance metrics
    portfolio_returns = returns_df.dot(weights)
    realized_var = historical_var(portfolio_returns)
    sharpe = calculate_sharpe_ratio(portfolio_returns)
    var_efficiency = abs(realized_var - var_limit) / var_limit
    
    return {
        'sharpe_ratio': sharpe,
        'var_efficiency': var_efficiency,
        'max_drawdown': calculate_max_drawdown(portfolio_returns),
        'turnover': calculate_turnover(weights)
    }

Au cours du test, il s'est avéré que le modèle fonctionnait particulièrement bien pendant les périodes de volatilité accrue. La composante VaR limite efficacement les risques, et l'optimisation Markowitz continue de rechercher des opportunités de rentabilité.

La version finale du système comprend également un mécanisme d’ajustement automatique des paramètres :

def auto_tune_parameters(returns_df, initial_params, optimization_window=252):
    best_params = initial_params
    best_score = float('-inf')
    
    for var_limit in np.arange(0.01, 0.05, 0.005):
        for shrinkage in np.arange(0.2, 0.8, 0.1):
            params = {'var_limit': var_limit, 'shrinkage': shrinkage}
            score = backtest_model(returns_df, params, optimization_window)
            
            if score > best_score:
                best_score = score
                best_params = params
                
    return best_params

Dans la section suivante, j'expliquerai comment ce modèle combiné est appliqué à la gestion dynamique des positions dans le cadre du trading réel.

Gestion dynamique de la taille des positions

La transposition du modèle théorique en un système de trading pratique a nécessité la résolution de nombreux problèmes techniques. Le principal enjeu était la gestion dynamique de la taille des positions, en tenant compte des conditions actuelles du marché et des pondérations optimales calculées pour le portefeuille.

Le système repose sur une classe dédiée à la gestion des positions :

class PositionManager:
    def __init__(self, account_balance, risk_limit=0.02):
        self.balance = account_balance
        self.risk_limit = risk_limit
        self.positions = {}
        
    def calculate_position_size(self, symbol, weight, var_estimate):
        symbol_info = mt5.symbol_info(symbol)
        pip_value = symbol_info.trade_tick_value * 10
        
        # Calculate the position size taking into account VaR
        max_risk_amount = self.balance * self.risk_limit * abs(weight)
        position_size = max_risk_amount / (abs(var_estimate) * pip_value)
        
        # Round to minimum lot
        return round(position_size / symbol_info.volume_step) * symbol_info.volume_step

Pour changer les positions en douceur, j'ai mis au point un mécanisme d'ordres partiels :

def adjust_positions(self, target_positions):
    for symbol, target_size in target_positions.items():
        current_size = self.get_current_position(symbol)
        if abs(target_size - current_size) > self.min_adjustment:
            # Break big changes into pieces
            steps = min(5, int(abs(target_size - current_size) / self.min_adjustment))
            step_size = (target_size - current_size) / steps
            
            for i in range(steps):
                next_size = current_size + step_size
                self.execute_order(symbol, next_size - current_size)
                current_size = next_size
                time.sleep(1)  # Prevent order flooding

J'ai accordé une attention particulière au contrôle des risques lors des modifications de positions :

def execute_order(self, symbol, size_delta, max_slippage=10):
    if size_delta > 0:
        order_type = mt5.ORDER_TYPE_BUY
    else:
        order_type = mt5.ORDER_TYPE_SELL
    
    # Get current prices
    tick = mt5.symbol_info_tick(symbol)
    
    # Set VaR-based stop loss
    if order_type == mt5.ORDER_TYPE_BUY:
        stop_loss = tick.bid - (self.var_estimates[symbol] * tick.bid)
        take_profit = tick.bid + (self.var_estimates[symbol] * 2 * tick.bid)
    else:
        stop_loss = tick.ask + (self.var_estimates[symbol] * tick.ask)
        take_profit = tick.ask - (self.var_estimates[symbol] * 2 * tick.ask)
    
    request = {
        "action": mt5.TRADE_ACTION_DEAL,
        "symbol": symbol,
        "volume": abs(size_delta),
        "type": order_type,
        "price": tick.ask if order_type == mt5.ORDER_TYPE_BUY else tick.bid,
        "sl": stop_loss,
        "tp": take_profit,
        "deviation": max_slippage,
        "magic": 234000,
        "comment": "var_based_adjustment",
        "type_time": mt5.ORDER_TIME_GTC,
        "type_filling": mt5.ORDER_FILLING_IOC,
    }
    
    result = mt5.order_send(request)
    return self.handle_order_result(result)

J'ai ajouté un système de surveillance de la volatilité pour me protéger contre les fortes fluctuations du marché :

def monitor_volatility(self, returns_df, threshold=2.0):
    # Current volatility calculation
    current_vol = returns_df.tail(20).std() * np.sqrt(252)
    historical_vol = returns_df.std() * np.sqrt(252)
    
    if current_vol > historical_vol * threshold:
        # Reduce positions in case of increased volatility
        self.reduce_exposure(current_vol / historical_vol)
        return False
        
    return True

Le système prévoit également une fermeture automatique des positions en cas de seuils de risque critiques :

def emergency_close(self, max_loss_percent=5.0):
    total_loss = sum(pos.profit for pos in mt5.positions_get())
    if total_loss < -self.balance * max_loss_percent / 100:
        print("Emergency closure triggered!")
        for position in mt5.positions_get():
            self.close_position(position.ticket)

Il en résulte un système robuste de gestion des positions, capable de fonctionner efficacement dans diverses conditions de marché. La section suivante portera sur le système de contrôle des risques basé sur la VaR.

Système de contrôle des risques de portefeuille

Après la mise en place d'une gestion dynamique des positions, j'ai été confronté à la nécessité de créer un système complet de contrôle des risques au niveau du portefeuille. L'expérience a montré que le contrôle local des risques liés aux positions individuelles n'est pas suffisant ; une approche globale est nécessaire.

J'ai commencé par la création d'une classe pour le suivi des risques de portefeuille :

class PortfolioRiskManager:
    def __init__(self, max_portfolio_var=0.03, max_correlation=0.7, max_drawdown=0.1):
        self.max_portfolio_var = max_portfolio_var
        self.max_correlation = max_correlation
        self.max_drawdown = max_drawdown
        self.current_drawdown = 0
        self.peak_balance = 0
        
    def update_portfolio_metrics(self, positions, returns_df):
        # Calculation of current portfolio weights
        total_exposure = sum(abs(pos.volume) for pos in positions)
        weights = {pos.symbol: pos.volume/total_exposure for pos in positions}
        
        # Update portfolio VaR
        self.current_var = self.calculate_portfolio_var(returns_df, weights)
        
        # Check correlations
        self.check_correlations(returns_df, weights)

J'ai accordé une attention particulière au suivi des corrélations entre les instruments :

def check_correlations(self, returns_df, weights):
    corr_matrix = returns_df.corr()
    high_corr_pairs = []
    
    for i in returns_df.columns:
        for j in returns_df.columns:
            if i < j and abs(corr_matrix.loc[i,j]) > self.max_correlation:
                if weights.get(i, 0) > 0 and weights.get(j, 0) > 0:
                    high_corr_pairs.append((i, j, corr_matrix.loc[i,j]))
                    
    if high_corr_pairs:
        self.handle_high_correlations(high_corr_pairs, weights)

J'ai mis en œuvre une gestion dynamique des risques en fonction des conditions du marché :

def adjust_risk_limits(self, market_state):
    volatility_factor = market_state.get('volatility_ratio', 1.0)
    trend_strength = market_state.get('trend_strength', 0.5)
    
    # Adapt limits to market conditions
    self.max_portfolio_var *= np.sqrt(volatility_factor)
    if trend_strength > 0.7:  # Strong trend
        self.max_drawdown *= 1.2  # Allow a big drawdown
    elif trend_strength < 0.3:  # Weak trend
        self.max_drawdown *= 0.8  # Reduce the acceptable drawdown

Le système de surveillance du drawdown s'est avéré particulièrement intéressant :

def monitor_drawdown(self, current_balance):
    if current_balance > self.peak_balance:
        self.peak_balance = current_balance
    
    self.current_drawdown = (self.peak_balance - current_balance) / self.peak_balance
    
    if self.current_drawdown > self.max_drawdown:
        return self.handle_excessive_drawdown()
    elif self.current_drawdown > self.max_drawdown * 0.8:
        return self.reduce_risk_exposure(0.8)
    
    return True

J'ai ajouté un système de tests de résistance pour me protéger contre les événements extrêmes :

def stress_test_portfolio(self, returns_df, weights, scenarios=1000):
    results = []
    
    for _ in range(scenarios):
        # Simulate extreme conditions
        stress_returns = returns_df.copy()
        
        # Increase volatility
        vol_multiplier = np.random.uniform(1.5, 3.0)
        stress_returns *= vol_multiplier
        
        # Add random shocks
        shock_magnitude = np.random.uniform(-0.05, 0.05)
        stress_returns += shock_magnitude
        
        # Calculate losses in a stress scenario
        portfolio_return = (stress_returns * weights).sum(axis=1)
        results.append(portfolio_return.min())
    
    return np.percentile(results, 1)  # 99% VaR in case of a stress

Il en résulte un système de protection du capital à plusieurs niveaux qui prévient efficacement les risques excessifs et aide à traverser les périodes de forte volatilité. Dans la section suivante, j'expliquerai comment tous ces éléments interagissent dans le cadre du trading réel.

Visualisation des résultats d'analyse

La visualisation est devenue une étape importante de mes recherches. Après avoir implémenté tous les modules de calcul, il a été nécessaire de créer une représentation visuelle des résultats. J'ai développé plusieurs composants graphiques clés qui permettent de surveiller les performances du système en temps réel.

J'ai commencé par visualiser la structure du portefeuille et son évolution :

def plot_portfolio_composition(weights_history):
    plt.figure(figsize=(15, 8))
    ax = plt.gca()
    
    # Create a graph of weight changes over time
    dates = weights_history.index
    bottom = np.zeros(len(dates))
    
    for symbol in weights_history.columns:
        plt.fill_between(dates, bottom, bottom + weights_history[symbol], 
                        label=symbol, alpha=0.6)
        bottom += weights_history[symbol]
    
    plt.title('Evolution of portfolio structure')
    plt.legend(bbox_to_anchor=(1.05, 1), loc='upper left')
    plt.grid(True, alpha=0.3)

Une attention particulière a été portée à la visualisation des risques. J'ai également développé une carte thermique de la VaR pour différentes paires de devises :

def plot_var_heatmap(var_matrix):
    plt.figure(figsize=(12, 8))
    sns.heatmap(var_matrix, annot=True, cmap='RdYlBu_r', 
                fmt='.2%', center=0)
    plt.title('Portfolio risk map (VaR)')
    
    # Add a timestamp
    plt.annotate(f'Last update: {datetime.now().strftime("%Y-%m-%d %H:%M")}',
                xy=(0.01, -0.1), xycoords='axes fraction')

Pour analyser la rentabilité, j'ai créé un graphique interactif mettant en évidence les événements importants :

def plot_performance_analytics(returns_df, var_values, significant_events):
    fig = plt.figure(figsize=(15, 10))
    gs = GridSpec(2, 1, height_ratios=[3, 1])
    
    # Returns graph
    ax1 = plt.subplot(gs[0])
    cumulative_returns = (1 + returns_df).cumprod()
    ax1.plot(cumulative_returns.index, cumulative_returns, 
             label='Portfolio returns')
    
    # Mark important events
    for date, event in significant_events.items():
        ax1.axvline(x=date, color='r', linestyle='--', alpha=0.3)
        ax1.annotate(event, xy=(date, ax1.get_ylim()[1]),
                    xytext=(10, 10), textcoords='offset points',
                    rotation=45)
    
    # VaR graph
    ax2 = plt.subplot(gs[1])
    ax2.fill_between(var_values.index, -var_values, 
                     color='lightblue', alpha=0.5,
                     label='Value at Risk')

J'ai ajouté un tableau de bord interactif pour suivre l'état du portefeuille :

class PortfolioDashboard:
    def __init__(self):
        self.fig = plt.figure(figsize=(15, 10))
        self.setup_subplots()
        
    def setup_subplots(self):
        gs = self.fig.add_gridspec(3, 2)
        self.ax_returns = self.fig.add_subplot(gs[0, :])
        self.ax_weights = self.fig.add_subplot(gs[1, 0])
        self.ax_risk = self.fig.add_subplot(gs[1, 1])
        self.ax_metrics = self.fig.add_subplot(gs[2, :])
        
    def update(self, portfolio_data):
        self._plot_returns(portfolio_data['returns'])
        self._plot_weights(portfolio_data['weights'])
        self._plot_risk_metrics(portfolio_data['risk'])
        self._update_metrics_table(portfolio_data['metrics'])
        
        plt.tight_layout()
        plt.show()

J'ai développé une visualisation dynamique pour analyser les corrélations :

def plot_correlation_dynamics(returns_df, window=60):
    # Calculation of dynamic correlations
    correlations = returns_df.rolling(window=window).corr()
    
    # Create an animated graph
    fig, ax = plt.subplots(figsize=(10, 10))
    
    def update(frame):
        ax.clear()
        sns.heatmap(correlations.loc[frame], 
                    vmin=-1, vmax=1, center=0,
                    cmap='RdBu', ax=ax)
        ax.set_title(f'Correlations on {frame.strftime("%Y-%m-%d")}')

Toutes ces visualisations permettent d'évaluer rapidement l'état du portefeuille et de prendre des décisions de trading. Dans la section suivante, je testerai le système.

Backtests de la stratégie

Une fois le développement de tous les composants du système achevé, je me suis trouvé confronté à la nécessité de procéder à des tests approfondis. Le processus s'est avéré beaucoup plus complexe que la simple analyse de données historiques. Il a fallu prendre en compte de nombreux facteurs : le slippage, les commissions et les spécificités d’exécution des ordres chez différents courtiers.

Les premiers backtests ont montré que l'approche classique avec des spreads fixes donne des résultats trop optimistes. Il était nécessaire de créer un modèle plus réaliste qui prenne en compte la variation des spreads en fonction de la volatilité et de l'heure de la journée.

J'ai accordé une attention particulière à la modélisation des gaps dans les données et des problèmes de liquidité. Dans la pratique, il arrive souvent qu'un ordre ne puisse pas être exécuté au prix estimé. Ces scénarios doivent être correctement gérés lors des tests.

Voici l'implémentation complète du système de backtesting :

class PortfolioBacktester:
    def __init__(self, initial_capital=100000, commission=0.0001):
        self.initial_capital = initial_capital
        self.commission = commission
        self.positions = {}
        self.trades_history = []
        self.balance_history = []
        self.var_history = []
        self.metrics = {}
        
    def run_backtest(self, returns_df, optimization_params):
        self.current_capital = self.initial_capital
        portfolio_returns = []
        
        # Preparing sliding windows for calculations
        window = 252  # Trading yesr
        for i in range(window, len(returns_df)):
            # Receive historical data for calculation
            historical_returns = returns_df.iloc[i-window:i]
            
            # Optimize the portfolio
            weights = self.optimize_portfolio(
                historical_returns, 
                optimization_params['target_return'],
                optimization_params['max_var']
            )
            
            # Calculate VaR for the current distribution
            current_var = self.calculate_portfolio_var(
                historical_returns, 
                weights,
                optimization_params['confidence_level']
            )
            
            # Check the need for rebalancing
            if self.should_rebalance(weights, current_var):
                self.execute_rebalancing(weights, returns_df.iloc[i])
            
            # Update positions and calculate profitability
            portfolio_return = self.update_positions(returns_df.iloc[i])
            portfolio_returns.append(portfolio_return)
            
            # Update metrics
            self.update_metrics(portfolio_return, current_var)
            
            # Check stop losses triggering
            self.check_stop_losses(returns_df.iloc[i])
        
        # Calculate the final metrics
        self.calculate_final_metrics(portfolio_returns)
        
    def optimize_portfolio(self, returns, target_return, max_var):
        # Using our hybrid optimization model
        opt = HybridOptimizer(returns, target_return, max_var)
        weights = opt.optimize()
        return self.apply_position_limits(weights)
        
    def execute_rebalancing(self, target_weights, current_prices):
        for symbol, target_weight in target_weights.items():
            current_weight = self.get_position_weight(symbol)
            if abs(target_weight - current_weight) > self.REBALANCING_THRESHOLD:
                # Simulate execution with slippage
                slippage = self.simulate_slippage(symbol, current_prices[symbol])
                trade_price = current_prices[symbol] * (1 + slippage)
                
                # Calculate the deal size
                trade_volume = self.calculate_trade_volume(
                    symbol, current_weight, target_weight
                )
                
                # Consider commissions
                commission = abs(trade_volume * trade_price * self.commission)
                self.current_capital -= commission
                
                # Set a deal to history
                self.record_trade(symbol, trade_volume, trade_price, commission)
                
    def update_metrics(self, portfolio_return, current_var):
        self.balance_history.append(self.current_capital)
        self.var_history.append(current_var)
        
        # Updating performance metrics
        self.metrics['max_drawdown'] = self.calculate_drawdown()
        self.metrics['sharpe_ratio'] = self.calculate_sharpe()
        self.metrics['var_efficiency'] = self.calculate_var_efficiency()
        
    def calculate_final_metrics(self, portfolio_returns):
        returns_series = pd.Series(portfolio_returns)
        
        self.metrics['total_return'] = (self.current_capital / self.initial_capital - 1)
        self.metrics['volatility'] = returns_series.std() * np.sqrt(252)
        self.metrics['sortino_ratio'] = self.calculate_sortino(returns_series)
        self.metrics['calmar_ratio'] = self.calculate_calmar()
        self.metrics['var_breaches'] = self.calculate_var_breaches()
        
    def simulate_slippage(self, symbol, price):
        # Simulate realistic slippage
        base_slippage = 0.0001  # Basic slippage
        time_factor = self.get_time_factor()  # Time dependency
        volume_factor = self.get_volume_factor(symbol)  # Volume dependency
        
        return base_slippage * time_factor * volume_factor

Les résultats des tests se sont révélés assez parlants. Le modèle hybride a démontré une résilience nettement supérieure aux chocs de marché par rapport aux approches classiques. Cela s’est vu tout particulièrement pendant les périodes de forte volatilité, lorsque la limite de la VaR protégeait efficacement le portefeuille contre les risques excessifs.

Débogage final

Après de nombreux mois de développement et de tests, je suis finalement parvenu à la version finale du système. Honnêtement, c'est très différent de ce que j'avais prévu au départ. La pratique a entraîné de nombreux changements, dont certains étaient tout à fait inattendus.

Le premier changement majeur a concerné ma façon de gérer les données. Je me suis rendu compte que tester le système uniquement sur des données historiques ne suffisait pas ; il était nécessaire de vérifier son comportement dans une grande variété de conditions de marché. J'ai donc développé un système de génération de données synthétiques. Cela paraît simple, mais en réalité, cela a pris plusieurs semaines.

J'ai commencé par diviser toutes les paires de devises en groupes en fonction de leur liquidité. Le premier groupe comprenait les paires majeures, telles que l'EUR/USD et le GBP/USD. Le deuxième regroupait des paires liées aux devises de matières premières, comme AUDUSD et USDCAD. Vinrent ensuite les paires croisées : EURJPY, GBPJPY et autres. Enfin, il y avait des paires exotiques, comme CADJPY et EURAUD. Pour chaque groupe, j'ai défini mes propres paramètres de volatilité et de corrélation, aussi proches que possible des valeurs réelles.

Mais le plus intéressant a commencé lorsque j'ai ajouté différents modes de marché. Imaginez : un tiers du temps, le marché est calme, la volatilité est faible. Un autre tiers correspond à des conditions de marché normales. Et le temps restant est synonyme de volatilité élevée, où tout s'emballe. J'ai également ajouté les tendances à long terme et les fluctuations cycliques. Il s'est avéré que c'était très similaire au marché réel.

L'optimisation du portefeuille a également nécessité certains efforts. Au départ, je pensais m'en sortir avec de simples restrictions sur le poids des positions, mais j'ai vite compris que ce n'était pas suffisant. J'ai donc ajouté des primes de risque dynamiques : plus la volatilité de la paire est élevée, plus le rendement potentiel devrait l'être. Restrictions introduites : minimum - 4% par position, maximum - 25%. Cela peut paraître beaucoup, mais si vous avez un effet de levier, c'est normal.

En parlant de levier. C'est une autre histoire. Au début, j'ai joué la carte de la sécurité et j'ai travaillé presque sans. Mais l'analyse a montré qu'un effet de levier modéré, d'environ 10 pour 1, améliorait considérablement les résultats. L'essentiel est de bien prendre en compte tous les coûts. Et il y en a un certain nombre : commissions de transaction (deux points de base), intérêts de maintien de l’effet de levier (0,01% par jour), glissement d’exécution. Tout cela devait être intégré à l'optimiseur.

Autre casse-tête : la protection contre les appels de marge. Après plusieurs expériences infructueuses, j'ai opté pour une solution simple : si le drawdown dépasse 10%, fermer toutes les positions et sauver au moins une partie du capital. Cela peut paraître conservateur, mais c'est très efficace sur le long terme.

Le plus difficile a été de faire le reporting. Lorsque votre système gère des dizaines de paires de devises et effectue constamment des achats et des ventes, il est tout simplement impossible de tout suivre. J'ai dû mettre en place tout un système de suivi : des rapports annuels avec une multitude d'indicateurs, des graphiques sur tout et n'importe quoi : de la simple valeur du portefeuille aux cartes thermiques de la répartition des pondérations.

J'ai mené les tests finaux sur une longue période, de 2000 à 2024. J'ai investi un million de dollars comme capital initial et je l'ai rééquilibré une fois par trimestre. J'étais satisfait des résultats. Le système s'adapte bien aux différentes conditions du marché et permet de maîtriser les risques. Même dans les crises les plus graves, il parvient à préserver l'essentiel de son capital.

Mais il reste encore beaucoup de travail à faire. J'aimerais ajouter l'apprentissage automatique pour prédire la volatilité. Actuellement, le système ne fonctionne qu'avec des données historiques. Je réfléchis également à la manière de rendre la gestion de l'effet de levier plus flexible. La fréquence de rééquilibrage pourrait également être optimisée. Parfois, un trimestre est trop long, alors qu’on peut aussi laisser les positions inchangées pendant six mois.

Le résultat a été complètement différent de ce que j'avais prévu au départ. Mais, comme on dit, le mieux est l'ennemi du bien. Le système fonctionne, maîtrise les risques et génère des profits. Et c'est là l'essentiel.

Conclusion

Ça a été une sacrée aventure. Lorsque j'ai commencé à m'intéresser à la théorie de Markowitz, je ne pouvais même pas imaginer ce que cela allait devenir. Je voulais simplement appliquer une approche conventionnelle au Forex. Mais au final, j'ai dû inventer une sorte de monstre de Frankenstein en combinant différentes approches de gestion des risques.

Le plus intéressant, c'est que j'ai réussi à croiser Markowitz avec la VaR, et ça marche vraiment ! Le plus drôle, c'est que les deux méthodes se sont révélées médiocres prises séparément, mais qu'ensemble, elles donnent d'excellents résultats. J’ai été particulièrement satisfait de la façon dont le système tient lorsque le marché est fortement secoué. La VaR est tout simplement incroyable comme contrainte d’optimisation.

Bien sûr, j'ai eu beaucoup de mal avec la partie technique. Mais désormais, tout est pris en compte : le slippage, les commissions et les modalités d'exécution.

J'ai testé le système sur des données historiques allant de 2000 à 2024. Les résultats étaient plutôt bons. Il s'adapte bien aux différentes conditions de marché et ne s'effondre même pas pendant les crises. Avec un effet de levier de 10 pour 1, le système tourne comme une horloge. L'essentiel est de contrôler rigoureusement les risques.

Mais il reste encore énormément de travail à faire. Je voudrais :

• ajouter l'apprentissage automatique aux prévisions de volatilité (ce sera le sujet du prochain article) ;
• régler la fréquence de rééquilibrage - peut-être peut-elle être optimisée ;
• rendre la gestion de l'effet de levier plus intelligente (l'effet de levier dynamique et l’utilisation dynamique et “intelligente” du capital seront également mis en œuvre dans de futurs articles) ;
• entraîner le système à s'adapter encore mieux aux différents régimes de marché.

En résumé, la principale conclusion est la suivante : un système de trading performant ne se résume pas à des équations tirées d'un manuel scolaire. Il faut ici comprendre le marché, maîtriser les technologies, et surtout savoir gérer les risques. Tous ces développements peuvent désormais être appliqués à d'autres marchés, et pas seulement au Forex. Bien qu'il y ait encore une marge de progression, les bases sont déjà en place et elles fonctionnent.