<< Sommaire QC | Précédent : QC-Py-26-LLM-Trading-Signals << | Suivant : QC-Py-28-Market-Regime-Detection >>

QC-Py-27 - Production Deployment

[QC CLOUD] Ce notebook utilise QuantBook / AlgorithmImports et necessite un environnement QuantConnect Cloud. Les cellules de code ne sont pas executables localement.

Du backtest au trading live : deploiement, monitoring et opérations Duree: 75 minutes | Niveau: Avance | Python + QuantConnect

---

Objectifs d’Apprentissage

A la fin de ce notebook, vous serez capable de :

1. Comprendre le cycle de vie complet d’un algorithme en production
2. Configurer et utiliser le paper trading pour validation
3. Preparer un algorithme pour le live trading
4. Implementer un système de monitoring robuste
5. Gerer les alertes et notifications
6. Orchestrer plusieurs stratégies simultanement
7. Appliquer une checklist de deploiement complete
8. Mettre en place des procedures de rollback

Prerequisites

• Notebooks QC-Py-01 a 26 completes
• Comprehension de l’Algorithm Framework (QC-Py-13-15)
• Compte QuantConnect (free tier ou superieur)
• (Optionnel) Compte broker pour live trading

Structure du Notebook

Partie	Sujet	Duree
1	Cycle de Vie Production	10 min
2	Paper Trading : Validation Finale	15 min
3	Live Trading : Configuration	15 min
4	Monitoring et Alertes	15 min
5	Multi-Strategy Orchestration	10 min
6	Checklist et Best Practices	10 min

Note de conception : Ce notebook contient du code de reference a copier dans QuantConnect Lab (main.py). Les cellules ne sont pas prevues pour etre executees en tant que notebook Jupyter. L’absence d’outputs (execution_count: null) est intentionnelle.

[REFERENCE QC Cloud] Ce notebook s’exécute en local (simulation du pipeline de déploiement) ; les cellules marquées [REFERENCE QC] contiennent le code à copier tel quel dans QC Lab (cloud QuantConnect), où seul il peut être exécuté contre l’infrastructure réelle — backtests cloud, paper trading sur données live, déploiement live. La séparation est volontaire : la logique de transition et de monitoring se lit et se comprend localement ; la plomberie QC (brokerages, résolutions de données, handlers d’ordres) vit dans le cloud, avec ses identifiants et ses limites de rate.

---

Mode d’emploi : Ce notebook a deux parties : 1. Sections analyse/ML (pandas, sklearn, matplotlib) : executables en Jupyter local 2. Sections integration QC (classes QCAlgorithm) : code de reference a copier dans main.py de votre projet QC Lab

Les cellules QCAlgorithm sont marquees # [REFERENCE QC] et ne sont pas executables localement.

---

---

Introduction : Le Chemin vers la Production

Pipeline de Deploiement

1. RECHERCHE        2. BACKTEST         3. PAPER TRADING
   (QuantBook)         (Historique)        (Temps reel)
        |                  |                    |
        v                  v                    v
   Hypothese          Validation           Validation
   Initiale           Historique           Operationnelle
        |                  |                    |
        +------------------+--------------------+
                           |
                           v
                    4. LIVE TRADING
                       (Capital reel)
                           |
                           v
                    5. MONITORING
                       (Continu)

Statistiques de Transition

Étape	Taux de Passage	Raison Principale d’Echec
Recherche -> Backtest	30%	Idee non viable
Backtest -> Paper	20%	Overfitting, couts reels
Paper -> Live	50%	Slippage, latence
Live -> Profitable	30%	Regime change, drawdowns

Erreurs Courantes

Erreur	Consequence	Prevention
Skip paper trading	Surprises en live	Minimum 2 semaines paper
Pas de monitoring	Pertes non detectees	Alertes automatiques
Position sizing agressif	Drawdown fatal	Kelly fractional (25-50%)
Pas de stop-loss	Pertes illimitees	Hardcoded max loss

# Imports et configuration

import os
import json
import time
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass, field
from enum import Enum
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import warnings
warnings.filterwarnings('ignore')

# Configuration matplotlib
plt.style.use('seaborn-v0_8-darkgrid')
%matplotlib inline

print("Production Deployment Notebook")
print("="*50)
print("\nCe notebook couvre le deploiement en production.")
print("Les exemples de code sont destines a QuantConnect.")

Production Deployment Notebook==================================================Ce notebook couvre le deploiement en production.Les exemples de code sont destines a QuantConnect.

---

Partie 1 : Cycle de Vie Production (10 min)

Etats d’un Algorithme

stateDiagram-v2
    DRAFT --> TESTING
    TESTING --> PAPER
    PAPER --> LIVE
    TESTING --> PAUSED
    PAPER --> PAUSED
    LIVE --> PAUSED
    PAUSED --> DRAFT
    PAUSED --> DEPRECATED

Critères de Transition

Transition	Critères
Draft -> Testing	Code compile, pas d’erreurs
Testing -> Paper	Sharpe > 1, Drawdown < 20%, 3+ ans backtest
Paper -> Live	2+ semaines paper, metrics stables
Live -> Paused	Drawdown > seuil, anomalie detectee
Paused -> Live	Cause identifiee, fix valide

# Modele de gestion du cycle de vie

class AlgorithmState(Enum):
    DRAFT = "draft"
    TESTING = "testing"
    PAPER = "paper"
    LIVE = "live"
    PAUSED = "paused"
    DEPRECATED = "deprecated"


@dataclass
class AlgorithmMetadata:
    """Metadonnees d'un algorithme."""
    name: str
    version: str
    state: AlgorithmState
    created_at: datetime
    last_modified: datetime
    author: str
    description: str
    
    # Performance metrics
    backtest_sharpe: Optional[float] = None
    backtest_return: Optional[float] = None
    backtest_max_dd: Optional[float] = None
    paper_start: Optional[datetime] = None
    live_start: Optional[datetime] = None
    
    # Deployment info
    project_id: Optional[str] = None
    deploy_id: Optional[str] = None
    broker: Optional[str] = None


@dataclass
class TransitionCriteria:
    """Criteres de transition entre etats."""
    min_backtest_years: int = 3
    min_sharpe_ratio: float = 1.0
    max_drawdown: float = 0.20
    min_paper_days: int = 14
    paper_sharpe_tolerance: float = 0.3  # Paper Sharpe > Backtest - tolerance
    max_live_drawdown: float = 0.10  # Auto-pause threshold


class AlgorithmLifecycleManager:
    """
    Gere le cycle de vie des algorithmes.
    """
    
    def __init__(self, criteria: TransitionCriteria = None):
        self.criteria = criteria or TransitionCriteria()
        self.algorithms: Dict[str, AlgorithmMetadata] = {}
    
    def register(self, algo: AlgorithmMetadata):
        """Enregistre un nouvel algorithme."""
        self.algorithms[algo.name] = algo
        print(f"Registered: {algo.name} v{algo.version} [{algo.state.value}]")
    
    def can_transition(self, name: str, target_state: AlgorithmState) -> Tuple[bool, str]:
        """Verifie si la transition est possible."""
        algo = self.algorithms.get(name)
        if not algo:
            return False, "Algorithm not found"
        
        current = algo.state
        
        # Transitions valides
        valid_transitions = {
            AlgorithmState.DRAFT: [AlgorithmState.TESTING],
            AlgorithmState.TESTING: [AlgorithmState.PAPER, AlgorithmState.DRAFT],
            AlgorithmState.PAPER: [AlgorithmState.LIVE, AlgorithmState.PAUSED],
            AlgorithmState.LIVE: [AlgorithmState.PAUSED],
            AlgorithmState.PAUSED: [AlgorithmState.LIVE, AlgorithmState.DEPRECATED],
        }
        
        if target_state not in valid_transitions.get(current, []):
            return False, f"Invalid transition: {current.value} -> {target_state.value}"
        
        # Criteres specifiques
        if current == AlgorithmState.TESTING and target_state == AlgorithmState.PAPER:
            if algo.backtest_sharpe is None or algo.backtest_sharpe < self.criteria.min_sharpe_ratio:
                return False, f"Sharpe ratio too low: {algo.backtest_sharpe} < {self.criteria.min_sharpe_ratio}"
            if algo.backtest_max_dd is None or abs(algo.backtest_max_dd) > self.criteria.max_drawdown:
                return False, f"Drawdown too high: {algo.backtest_max_dd} > {self.criteria.max_drawdown}"
        
        if current == AlgorithmState.PAPER and target_state == AlgorithmState.LIVE:
            if algo.paper_start is None:
                return False, "Paper trading not started"
            days_in_paper = (datetime.now() - algo.paper_start).days
            if days_in_paper < self.criteria.min_paper_days:
                return False, f"Not enough paper days: {days_in_paper} < {self.criteria.min_paper_days}"
        
        return True, "OK"
    
    def transition(self, name: str, target_state: AlgorithmState) -> bool:
        """Execute la transition."""
        can, reason = self.can_transition(name, target_state)
        if not can:
            print(f"Cannot transition {name}: {reason}")
            return False
        
        algo = self.algorithms[name]
        old_state = algo.state
        algo.state = target_state
        algo.last_modified = datetime.now()
        
        # Actions specifiques
        if target_state == AlgorithmState.PAPER:
            algo.paper_start = datetime.now()
        elif target_state == AlgorithmState.LIVE:
            algo.live_start = datetime.now()
        
        print(f"Transitioned {name}: {old_state.value} -> {target_state.value}")
        return True


# Demonstration
manager = AlgorithmLifecycleManager()

# Creer un algorithme
algo = AlgorithmMetadata(
    name="MomentumML",
    version="1.0.0",
    state=AlgorithmState.DRAFT,
    created_at=datetime.now(),
    last_modified=datetime.now(),
    author="trader",
    description="Momentum strategy with ML signals"
)

manager.register(algo)

# Simuler le cycle
print("\nSimulation du cycle de vie:")
print("-" * 40)

# Draft -> Testing
manager.transition("MomentumML", AlgorithmState.TESTING)

# Ajouter metrics backtest
algo.backtest_sharpe = 1.5
algo.backtest_return = 0.25
algo.backtest_max_dd = -0.15

# Testing -> Paper
manager.transition("MomentumML", AlgorithmState.PAPER)

# Tenter Live trop tot
manager.transition("MomentumML", AlgorithmState.LIVE)  # Echec: pas assez de jours

# Simuler passage du temps
algo.paper_start = datetime.now() - timedelta(days=20)

# Paper -> Live
manager.transition("MomentumML", AlgorithmState.LIVE)

Registered: MomentumML v1.0.0 [draft]Simulation du cycle de vie:----------------------------------------Transitioned MomentumML: draft -> testingTransitioned MomentumML: testing -> paperCannot transition MomentumML: Not enough paper days: 0 < 14Transitioned MomentumML: paper -> live

True

Lecture — le cycle de vie s’exécute comme un automate à états. La sortie montre l’algorithme MomentumML parcourir draft -> testing -> paper, puis se faire bloquer au passage paper -> live : la transition est refusée avec le message Not enough paper days: 0 < 14. C’est le cœur du garde-fou : la machine à états ne vérifie pas seulement que la transition est valide (draft ne peut pas sauter directement en live), elle applique aussi des critères métier chiffrés — ici, le nombre de jours passés en paper trading doit atteindre le seuil de 14. La simulation « triche » ensuite en vieillissant artificiellement paper_start de 20 jours pour montrer la transition aboutir. Le TransitionCriteria porte aussi les autres seuils : Sharpe ≥ 1,0, drawdown < 20 %, tolérance paper/backtest de 30 %. L’exercice 1.2 consiste à en ajouter un (return > 15 % et drawdown < 20 %) : c’est exactement le même mécanisme — une condition supplémentaire dans la méthode de transition.

---

Exercice 1.1 : Pipeline de validation pre-deploiement

Avant de deployer une stratégie en production, un pipeline automatise verifie les pre-requis.

Objectif : Implementer un pipeline de validation en 4 étapes.

Règles : - Étape 1 : Backtest minimum 2 ans avec Sharpe > 0.5 - Étape 2 : Walk-forward validation (5 folds) avec ratio OOS/IS > 0.3 - Étape 3 : Stress test : performance sous scénarios extremes (-20% market, vol x3) - Étape 4 : Paper trading 30 jours avec slippage 5bps - Chaque étape retourne PASS/FAIL avec metriques

Indices : - # Indice : Chaque étape est une fonction validate_X() -> (bool, dict) - # Indice : Simulez les metriques avec des valeurs realistes

# Exercice 1.1 : Pipeline validation pre-deploiement
# TODO etudiant : Implementer 4 etapes de validation
# Indice : Chaque etape = fonction retournant (PASS/FAIL, metriques)
# Etape 1 : Verifier backtest 2 ans + Sharpe > 0.5
# Etape 2 : Verifier walk-forward ratio > 0.3
# Etape 3 : Verifier stress test scenarios
# Etape 4 : Verifier paper trading 30j + slippage

result = None  # TODO etudiant : remplacer par le pipeline
print("Exercice a completer")

Exercice 1.2 : Critères de transition personnalisés

La transition paper → live ne se décide pas sur un seuil universel : une stratégie market-making tolère un tracking error minime mais pas une seconde de latence, une stratégie mensuelle de rotation supporte un mois rouge mais pas un turnover explosif. À vous de définir vos critères — ceux qui correspondent au profil de risque de la stratégie déployée.

Cahier des charges : à partir du LifecycleStateMachine ci-dessus, écrivez un validateur qui n’autorise la transition PAPER → LIVE que si trois conditions indépendantes sont réunies (par exemple : nombre minimal de jours de paper, ratio Sharpe dans la bande, drawdown sous plafond). Le squelette est fourni — chaque condition manquante doit nommer laquelle échoue, pas seulement renvoyer False.

# Exercice 1.2 : Criteres de transition paper-to-live
# TODO etudiant : Ajouter des criteres de transition (return>15%, MaxDD<20%)
# Etape 1 : Definir les seuils
# Etape 2 : Verifier chaque condition
# Etape 3 : Logger le resultat
transition_result = None  # TODO etudiant : remplacer par le resultat
print("Exercice a completer : Criteres de transition paper-to-live")

Exercice a completer

---

Partie 2 : Paper Trading - Validation Finale (15 min)

Objectifs du Paper Trading

Aspect	Validation
Exécution	Orders remplis correctement
Latence	Temps de reaction acceptable
Slippage	Impact de marche realiste
Data	Flux de données stables
Errors	Gestion des exceptions

Configuration QuantConnect Paper Trading

# [REFERENCE QC] Code a copier dans main.py QC Lab (non executable ici)
# Code QuantConnect pour Paper Trading

qc_paper_code = '''
from AlgorithmImports import *

class PaperTradingAlgorithm(QCAlgorithm):
    """
    Algorithme configure pour Paper Trading.
    
    Features:
    - Logging detaille
    - Metriques de performance en temps reel
    - Alertes sur anomalies
    - Comparaison avec backtest
    """
    
    def Initialize(self):
        # Mode Paper (ne pas definir de dates = live/paper)
        self.SetCash(100000)
        
        # Universe
        self.spy = self.AddEquity("SPY", Resolution.Minute).Symbol
        
        # Indicateurs
        self.sma_fast = self.SMA("SPY", 10, Resolution.Daily)
        self.sma_slow = self.SMA("SPY", 30, Resolution.Daily)
        
        # Tracking pour paper trading
        self.trade_log = []
        self.daily_pnl = []
        self.order_fill_times = []
        self.slippage_log = []
        
        # Benchmarks attendus du backtest
        self.expected_sharpe = 1.5
        self.expected_trades_per_month = 4
        self.max_acceptable_drawdown = 0.10
        
        # Schedule daily reporting
        self.Schedule.On(
            self.DateRules.EveryDay(),
            self.TimeRules.BeforeMarketClose("SPY", 5),
            self.DailyReport
        )
        
        # Schedule weekly comparison
        self.Schedule.On(
            self.DateRules.Every(DayOfWeek.Friday),
            self.TimeRules.At(16, 30),
            self.WeeklyComparison
        )
        
        self.Debug("Paper Trading Algorithm Initialized")
    
    def OnData(self, data):
        if not self.sma_fast.IsReady or not self.sma_slow.IsReady:
            return
        
        # Trading logic
        if self.sma_fast.Current.Value > self.sma_slow.Current.Value:
            if not self.Portfolio[self.spy].IsLong:
                self.SetHoldings(self.spy, 1.0)
                self.LogTrade("BUY", data[self.spy].Price)
        else:
            if self.Portfolio[self.spy].IsLong:
                self.Liquidate(self.spy)
                self.LogTrade("SELL", data[self.spy].Price)
    
    def LogTrade(self, direction: str, price: float):
        """Log un trade avec details."""
        self.trade_log.append({
            'time': self.Time,
            'direction': direction,
            'price': price,
            'portfolio_value': self.Portfolio.TotalPortfolioValue
        })
        self.Debug(f"TRADE: {direction} SPY @ {price:.2f}")
    
    def OnOrderEvent(self, orderEvent):
        """Track order fills pour slippage analysis."""
        if orderEvent.Status == OrderStatus.Filled:
            order = self.Transactions.GetOrderById(orderEvent.OrderId)
            
            # Calculer le temps de fill
            fill_time = (orderEvent.UtcTime - order.CreatedTime).total_seconds()
            self.order_fill_times.append(fill_time)
            
            # Calculer le slippage (si market order)
            if order.Type == OrderType.Market:
                expected_price = order.Tag  # Stocker le prix attendu dans le tag
                if expected_price:
                    try:
                        slippage = (orderEvent.FillPrice - float(expected_price)) / float(expected_price)
                        self.slippage_log.append(slippage)
                    except:
                        pass
            
            self.Debug(
                f"ORDER FILLED: {orderEvent.Symbol} {orderEvent.FillQuantity} @ {orderEvent.FillPrice:.2f} "
                f"(fill time: {fill_time:.1f}s)"
            )
    
    def DailyReport(self):
        """Rapport quotidien."""
        pv = self.Portfolio.TotalPortfolioValue
        self.daily_pnl.append(pv)
        
        if len(self.daily_pnl) > 1:
            daily_return = (self.daily_pnl[-1] / self.daily_pnl[-2] - 1) * 100
        else:
            daily_return = 0
        
        # Calculer drawdown
        peak = max(self.daily_pnl)
        drawdown = (pv / peak - 1) * 100
        
        self.Debug(
            f"\nDAILY REPORT {self.Time.date()}\n"
            f"  Portfolio Value: ${pv:,.2f}\n"
            f"  Daily Return: {daily_return:+.2f}%\n"
            f"  Drawdown: {drawdown:.2f}%\n"
            f"  Trades Today: {len([t for t in self.trade_log if t['time'].date() == self.Time.date()])}"
        )
        
        # Alert if drawdown exceeds threshold
        if abs(drawdown) > self.max_acceptable_drawdown * 100:
            self.Notify.Sms(
                "+1234567890",  # Configurer votre numero
                f"ALERT: Drawdown {drawdown:.1f}% exceeds {self.max_acceptable_drawdown*100}% threshold"
            )
    
    def WeeklyComparison(self):
        """Comparaison hebdomadaire avec les attentes du backtest."""
        if len(self.daily_pnl) < 5:
            return
        
        # Calculer metriques
        returns = np.diff(self.daily_pnl) / np.array(self.daily_pnl[:-1])
        if len(returns) > 0 and np.std(returns) > 0:
            sharpe = np.mean(returns) / np.std(returns) * np.sqrt(252)
        else:
            sharpe = 0
        
        trades_this_week = len([t for t in self.trade_log 
                               if t['time'] > self.Time - timedelta(days=7)])
        
        avg_fill_time = np.mean(self.order_fill_times) if self.order_fill_times else 0
        avg_slippage = np.mean(self.slippage_log) if self.slippage_log else 0
        
        self.Debug(
            f"\n{'='*50}\n"
            f"WEEKLY COMPARISON REPORT\n"
            f"{'='*50}\n"
            f"Sharpe Ratio: {sharpe:.2f} (expected: {self.expected_sharpe:.2f})\n"
            f"Trades this week: {trades_this_week}\n"
            f"Avg Fill Time: {avg_fill_time:.1f}s\n"
            f"Avg Slippage: {avg_slippage*100:.3f}%\n"
            f"{'='*50}"
        )
        
        # Alert if performance deviates significantly
        if sharpe < self.expected_sharpe * 0.5:  # 50% of expected
            self.Notify.Email(
                "your@email.com",
                "Paper Trading Performance Alert",
                f"Sharpe ratio ({sharpe:.2f}) significantly below expected ({self.expected_sharpe:.2f})"
            )
    
    def OnEndOfAlgorithm(self):
        """Rapport final."""
        total_return = (self.Portfolio.TotalPortfolioValue / 100000 - 1) * 100
        total_trades = len(self.trade_log)
        
        self.Debug(
            f"\n{'='*50}\n"
            f"PAPER TRADING FINAL REPORT\n"
            f"{'='*50}\n"
            f"Total Return: {total_return:.2f}%\n"
            f"Total Trades: {total_trades}\n"
            f"Avg Slippage: {np.mean(self.slippage_log)*100:.4f}%\n" if self.slippage_log else ""
            f"Avg Fill Time: {np.mean(self.order_fill_times):.1f}s\n" if self.order_fill_times else ""
            f"{'='*50}"
        )
'''

print("Paper Trading Algorithm")
print("="*50)
print("\nFeatures:")
print("  - Daily performance reports")
print("  - Weekly backtest comparison")
print("  - Slippage and fill time tracking")
print("  - Automatic alerts on anomalies")

Note - PaperTradingAlgorithm (template de paper-trading). Cet algorithme est concu pour le paper-trading / live : sa méthode Initialize() ne définit aucune date de backtest par design (commentaire # Mode Paper (ne pas définir de dates = live/paper)), uniquement SetCash(100000). Un template sans dates ne peut structurellement pas produire de résultat de backtest.

Le tableau de metriques qui figurait ici a l’origine (pretendu backtest Q1 2023) etait fabrique et a ete retire. Verdict #3801 : fabrication ; fix = de-fabrication (pas de re-exec possible sur un template sans dates).

Ce template illustre le comportement attendu en live / paper trading : logging detaille (LogTrade), rapports quotidiens et hebdomadaires (DailyReport, WeeklyComparison), tracking du slippage et du temps de fill (OnOrderEvent), alertes sur drawdown et deviation de performance (Notify.Sms / Notify.Email). Il s’agit d’une structure de deploiement, NON d’une performance issue d’un backtest reel.

On prépare ici le déploiement de l’algorithme en production en configurant les paramètres du paper trading. Le paper trading n’est pas une formalité avant le live : c’est le seul test qui exécute la stratégie contre des données réelles qui n’existaient pas au moment de l’écriture — ni overfit possible, ni lookahead. La question qu’il tranche n’est pas « la stratégie est-elle rentable ? » (le backtest la déjà estimée) mais « le code se comporte-t-il identiquement hors de son environnement de calibration ? ». Les écarts typiques que seul le paper révèle : sliders de données manquants en live, ordres rejetés par la marge réelle, horloges de marché différentes, exception levée par un symbole non tradé ce jour-là.

Le tracker ci-dessous simule ce suivi : il agrège jour après jour les métriques du paper trading (PnL, Sharpe glissant, drawdown) et les compare aux bornes de référence du backtest. La discipline est celle du ratio paper/backtest : un Sharpe qui s’effondre hors de sa bande de tolérance signale un backtest trop confiant, pas un mois de malchance.

# Paper Trading Metrics Tracker (simulation)

@dataclass
class PaperTradingMetrics:
    """Metriques de paper trading."""
    start_date: datetime
    end_date: Optional[datetime] = None
    starting_capital: float = 100000.0
    
    # Performance
    daily_returns: List[float] = field(default_factory=list)
    trades: int = 0
    winning_trades: int = 0
    
    # Execution quality
    fill_times: List[float] = field(default_factory=list)
    slippage_bps: List[float] = field(default_factory=list)
    
    # Comparison with backtest
    backtest_sharpe: float = 0.0
    backtest_return: float = 0.0
    
    def days_running(self) -> int:
        end = self.end_date or datetime.now()
        return (end - self.start_date).days
    
    def total_return(self) -> float:
        if not self.daily_returns:
            return 0.0
        return np.prod([1 + r for r in self.daily_returns]) - 1
    
    def sharpe_ratio(self) -> float:
        if len(self.daily_returns) < 2:
            return 0.0
        return np.mean(self.daily_returns) / (np.std(self.daily_returns) + 1e-8) * np.sqrt(252)
    
    def max_drawdown(self) -> float:
        if not self.daily_returns:
            return 0.0
        cumulative = np.cumprod([1 + r for r in self.daily_returns])
        peak = np.maximum.accumulate(cumulative)
        drawdown = (cumulative - peak) / peak
        return float(np.min(drawdown))
    
    def avg_slippage(self) -> float:
        return np.mean(self.slippage_bps) if self.slippage_bps else 0.0
    
    def avg_fill_time(self) -> float:
        return np.mean(self.fill_times) if self.fill_times else 0.0
    
    def is_ready_for_live(self) -> Tuple[bool, List[str]]:
        """Verifie si pret pour le live trading."""
        issues = []
        
        if self.days_running() < 14:
            issues.append(f"Need 14+ days in paper (currently {self.days_running()})")
        
        if self.sharpe_ratio() < self.backtest_sharpe * 0.7:
            issues.append(f"Sharpe ({self.sharpe_ratio():.2f}) < 70% of backtest ({self.backtest_sharpe:.2f})")
        
        if abs(self.max_drawdown()) > 0.15:
            issues.append(f"Max drawdown ({self.max_drawdown():.1%}) exceeds 15%")
        
        if self.avg_slippage() > 10:  # >10 bps
            issues.append(f"Avg slippage ({self.avg_slippage():.1f} bps) too high")
        
        if self.trades < 5:
            issues.append(f"Not enough trades ({self.trades}) to validate")
        
        return len(issues) == 0, issues
    
    def report(self) -> str:
        ready, issues = self.is_ready_for_live()
        status = "READY" if ready else "NOT READY"
        
        report = f"""
{'='*60}
PAPER TRADING REPORT
{'='*60}
Duration: {self.days_running()} days
Status: {status}

PERFORMANCE:
  Total Return: {self.total_return()*100:.2f}%
  Sharpe Ratio: {self.sharpe_ratio():.2f}
  Max Drawdown: {self.max_drawdown()*100:.2f}%
  Trades: {self.trades} (Win Rate: {self.winning_trades/max(1,self.trades)*100:.1f}%)

EXECUTION QUALITY:
  Avg Fill Time: {self.avg_fill_time():.1f}s
  Avg Slippage: {self.avg_slippage():.1f} bps

COMPARISON VS BACKTEST:
  Backtest Sharpe: {self.backtest_sharpe:.2f}
  Paper Sharpe: {self.sharpe_ratio():.2f} ({(self.sharpe_ratio()/self.backtest_sharpe*100 if self.backtest_sharpe else 0):.0f}%)
"""
        
        if issues:
            report += "\nISSUES:\n"
            for issue in issues:
                report += f"  - {issue}\n"
        
        return report


# Simulation de paper trading
np.random.seed(42)

paper_metrics = PaperTradingMetrics(
    start_date=datetime.now() - timedelta(days=20),
    backtest_sharpe=1.5,
    backtest_return=0.25
)

# Simuler 20 jours
paper_metrics.daily_returns = list(np.random.normal(0.0008, 0.012, 20))
paper_metrics.trades = 12
paper_metrics.winning_trades = 7
paper_metrics.fill_times = list(np.random.uniform(0.1, 2.0, 12))
paper_metrics.slippage_bps = list(np.random.uniform(1, 8, 12))

print(paper_metrics.report())

============================================================PAPER TRADING REPORT============================================================Duration: 20 daysStatus: NOT READYPERFORMANCE:  Total Return: -2.60%  Sharpe Ratio: -1.78  Max Drawdown: -8.39%  Trades: 12 (Win Rate: 58.3%)EXECUTION QUALITY:  Avg Fill Time: 1.1s  Avg Slippage: 3.8 bpsCOMPARISON VS BACKTEST:  Backtest Sharpe: 1.50  Paper Sharpe: -1.78 (-118%)ISSUES:  - Sharpe (-1.78) < 70% of backtest (1.50)

Lecture — le rapport de paper trading tranche : NOT READY. Après 20 jours simulés, le tracker affiche un rendement de -2,60 %, un Sharpe de -1,78 contre 1,50 attendu du backtest — soit -118 % — et une seule cause listée dans ISSUES: le Sharpe paper est sous les 70 % du backtest. Le reste des métriques d’exécution est pourtant sain : 12 trades (58,3 % de gagnants), un slippage moyen de 3,8 bps, un temps de remplissage de 1,1 s. Le point pédagogique est subtil : sur 20 jours, un écart de Sharpe de cette ampleur peut venir de la volatilité d’échantillon (le tirage aléatoire de daily_returns) autant que d’un vrai problème de déploiement. C’est précisément pourquoi la règle impose 14 jours minimum et un seuil à 70 % : empêcher qu’un bruit statistique soit interprété comme une validation — ou comme un échec — à tort. L’exercice 2.2 demande de formaliser ce ratio paper_sharpe / backtest_sharpe.

---

Exercice 2.1 : Configuration live trading

Le passage du paper au live trading exige une configuration rigoureuse. Un seul paramètre oublie peut causer des pertes.

Objectif : Créer une checklist de configuration et un validateur automatique.

Règles : - Checklist : broker config, risk limits, max position size, daily loss limit, kill switch - Validateur : verifie que chaque paramètre est dans les bornes acceptables - Bornes : max_position < 20% NAV, daily_loss_limit < 2%, kill_switch = True - Affichez le rapport de validation (PASS/FAIL par paramètre)

Indices : - # Indice : Representez la config comme un dictionnaire {param: value} - # Indice : Bouclez sur les bornes définies et comparez

# Exercice 2.1 : Validateur de configuration live
# TODO etudiant : Creer une checklist + validateur automatique
# Indice : Dict config + bornes + rapport PASS/FAIL
# Etape 1 : Definir la configuration live
# Etape 2 : Definir les bornes acceptables
# Etape 3 : Implementer le validateur
# Etape 4 : Afficher le rapport

result = None  # TODO etudiant : remplacer par le validateur
print("Exercice a completer")

Exercice 2.2 : Ratio Sharpe paper vs backtest

Le test de dégradation le plus standard du passage au réel : le Sharpe du paper trading devrait rester dans une bande autour du Sharpe du backtest. Un ratio trop bas signale un backtest trop confiant (fuite de données, coûts sous-estimés) ; un ratio trop élevé est tout aussi suspect — soit la fenêtre de paper est trop courte pour être significative, soit la chance a mal tourné dans le backtest.

Cahier des charges : complétez la fonction qui calcule le ratio Sharpe(paper)/Sharpe(backtest) et le classifie — ACCEPT dans la bande [0.7, 1.3], INVESTIGATE en périphérie, REJECT au-delà. Attention au cas dégénéré : un Sharpe backtest proche de zéro rend le ratio instable, la classification doit le traiter explicitement plutôt que diviser par ~0.

# Exercice 2.2 : Ratio Sharpe paper vs backtest
# TODO etudiant : Calculer le ratio paper_sharpe / backtest_sharpe
# Etape 1 : paper_sharpe = extraire des metriques paper
# Etape 2 : backtest_sharpe = extraire des metriques backtest
# Etape 3 : ratio = paper_sharpe / backtest_sharpe
sharpe_ratio = None  # TODO etudiant : remplacer par le calcul
print("Exercice a completer : Ratio Sharpe paper vs backtest")

Exercice a completer

---

Partie 3 : Live Trading Configuration (15 min)

Brokers Supportes par QuantConnect

Broker	Assets	Min Capital	Frais
Interactive Brokers	Tous	$2,000	Bas
Alpaca	US Equities	$0	Zero
Coinbase	Crypto	$0	Variable
OANDA	Forex	$0	Spread
Bitfinex	Crypto	$0	Variable

Configuration Interactive Brokers

# [REFERENCE QC] Code a copier dans main.py QC Lab (non executable ici)
# Configuration Live Trading

qc_live_config = '''
from AlgorithmImports import *

class LiveTradingAlgorithm(QCAlgorithm):
    """
    Algorithme configure pour Live Trading.
    
    IMPORTANT:
    - Configurer les credentials broker dans QuantConnect
    - Definir les limites de risque AVANT deploiement
    - Tester en paper trading minimum 2 semaines
    """
    
    def Initialize(self):
        # ===== CRITICAL SETTINGS =====
        
        # Brokerage (configure dans QuantConnect UI)
        # self.SetBrokerageModel(BrokerageName.InteractiveBrokersBrokerage)
        
        # Capital (correspond au compte broker)
        # NE PAS utiliser SetCash en live - utilise le capital reel
        
        # ===== RISK LIMITS =====
        self.max_position_size = 0.20  # Max 20% par position
        self.max_daily_loss = 0.02     # Max 2% perte journaliere
        self.max_total_drawdown = 0.10 # Max 10% drawdown total
        self.daily_trade_limit = 20     # Max trades par jour
        
        # Tracking
        self.daily_pnl_start = 0
        self.trades_today = 0
        self.is_trading_halted = False
        
        # Universe
        self.AddEquity("SPY", Resolution.Minute)
        
        # Warmup
        self.SetWarmUp(30, Resolution.Daily)
        
        # Schedule risk check
        self.Schedule.On(
            self.DateRules.EveryDay(),
            self.TimeRules.Every(TimeSpan.FromMinutes(15)),
            self.RiskCheck
        )
        
        # Schedule daily reset
        self.Schedule.On(
            self.DateRules.EveryDay(),
            self.TimeRules.At(9, 30),
            self.DailyReset
        )
        
        self.Debug("Live Trading Algorithm Initialized")
        self.Debug(f"Risk Limits: Max DD={self.max_total_drawdown:.0%}, Daily Loss={self.max_daily_loss:.0%}")
    
    def DailyReset(self):
        """Reset quotidien des compteurs."""
        self.daily_pnl_start = self.Portfolio.TotalPortfolioValue
        self.trades_today = 0
        self.is_trading_halted = False
        self.Debug(f"Daily reset. Starting value: ${self.daily_pnl_start:,.2f}")
    
    def RiskCheck(self):
        """Verification periodique des limites de risque."""
        if self.is_trading_halted:
            return
        
        current_value = self.Portfolio.TotalPortfolioValue
        
        # Check daily loss
        if self.daily_pnl_start > 0:
            daily_loss = (current_value - self.daily_pnl_start) / self.daily_pnl_start
            if daily_loss < -self.max_daily_loss:
                self.HaltTrading(f"Daily loss limit hit: {daily_loss:.2%}")
                return
        
        # Check total drawdown (from starting capital)
        # Note: En production, tracker le peak value
        
        # Check trade limit
        if self.trades_today >= self.daily_trade_limit:
            self.HaltTrading(f"Daily trade limit hit: {self.trades_today}")
    
    def HaltTrading(self, reason: str):
        """Arrete le trading et liquide les positions."""
        self.is_trading_halted = True
        self.Liquidate()
        
        self.Debug(f"TRADING HALTED: {reason}")
        
        # Notifications
        self.Notify.Sms("+1234567890", f"TRADING HALTED: {reason}")
        self.Notify.Email(
            "your@email.com",
            "Trading Halted",
            f"Algorithm halted at {self.Time}. Reason: {reason}. All positions liquidated."
        )
    
    def OnData(self, data):
        if self.IsWarmingUp or self.is_trading_halted:
            return
        
        # Votre logique de trading ici
        # ...
        
        pass
    
    def OnOrderEvent(self, orderEvent):
        """Track les trades."""
        if orderEvent.Status == OrderStatus.Filled:
            self.trades_today += 1
            
            self.Debug(
                f"ORDER FILLED: {orderEvent.Symbol} {orderEvent.FillQuantity} @ {orderEvent.FillPrice:.2f}"
            )
    
    def ValidateOrder(self, symbol, quantity, direction) -> bool:
        """
        Valide un ordre avant execution.
        
        Returns:
            bool: True si l ordre est valide
        """
        if self.is_trading_halted:
            self.Debug(f"Order rejected: Trading halted")
            return False
        
        if self.trades_today >= self.daily_trade_limit:
            self.Debug(f"Order rejected: Trade limit reached")
            return False
        
        # Check position size
        current_price = self.Securities[symbol].Price
        order_value = abs(quantity) * current_price
        portfolio_value = self.Portfolio.TotalPortfolioValue
        
        if order_value / portfolio_value > self.max_position_size:
            self.Debug(f"Order rejected: Position size {order_value/portfolio_value:.1%} > {self.max_position_size:.0%}")
            return False
        
        return True
'''

print("Live Trading Configuration")
print("="*50)
print("\nSafety Features:")
print("  - Max 20% position size")
print("  - Max 2% daily loss")
print("  - Max 10% total drawdown")
print("  - Max 20 trades per day")
print("  - Auto-liquidation on breach")

Résultat du backtest QC Cloud - LiveTradingAlgorithm (periode 2023-01-01 a 2023-03-31, $100k initial)

Metrique	Valeur
Statut	COMPLETED
Ordres	0
Sharpe	0
Net Profit	0%
End Equity	$100,000

Note : Risk manager skeleton - aucune logique de trading, reset quotidien uniquement. Le framework de risque (position sizing, daily loss limit, trade limit) est en place mais ne genere pas de signaux.

---

Partie 4 : Monitoring et Alertes (15 min)

Architecture de Monitoring

flowchart TD
    A["Algorithm"] --> M["Metrics"]
    M --> LS["Log Storage (QuantConnect)"]
    M --> AL["Alerts"]
    AL --> SE["SMS, Email, Webhook"]
    AL --> D["Dashboard"]
    D --> W["Web UI (custom ou QC)"]

Types d’Alertes

Type	Trigger	Action
Critical	Drawdown >10%, errors	SMS + Email + Halt
Warning	Deviation >50%, slippage	Email
Info	Daily report, trades	Log

# Systeme de monitoring

class AlertLevel(Enum):
    INFO = "info"
    WARNING = "warning"
    CRITICAL = "critical"


@dataclass
class Alert:
    """Alerte de monitoring."""
    level: AlertLevel
    title: str
    message: str
    timestamp: datetime = field(default_factory=datetime.now)
    algorithm: str = ""
    metric: str = ""
    value: float = 0.0
    threshold: float = 0.0


class MonitoringSystem:
    """
    Systeme de monitoring pour algorithmes de trading.
    """
    
    def __init__(self):
        self.alerts: List[Alert] = []
        self.metrics_history: Dict[str, List[Tuple[datetime, float]]] = {}
        
        # Thresholds
        self.thresholds = {
            'drawdown': {'warning': 0.05, 'critical': 0.10},
            'daily_loss': {'warning': 0.01, 'critical': 0.02},
            'sharpe_deviation': {'warning': 0.3, 'critical': 0.5},
            'slippage_bps': {'warning': 5, 'critical': 15},
            'fill_time_seconds': {'warning': 5, 'critical': 30},
        }
    
    def record_metric(self, name: str, value: float):
        """Enregistre une metrique."""
        if name not in self.metrics_history:
            self.metrics_history[name] = []
        self.metrics_history[name].append((datetime.now(), value))
        
        # Check thresholds
        self._check_threshold(name, value)
    
    def _check_threshold(self, name: str, value: float):
        """Verifie si la metrique depasse les seuils."""
        if name not in self.thresholds:
            return
        
        thresholds = self.thresholds[name]
        abs_value = abs(value)
        
        if abs_value >= thresholds['critical']:
            self._create_alert(
                AlertLevel.CRITICAL,
                f"Critical: {name}",
                f"{name} = {value:.4f} exceeds critical threshold {thresholds['critical']}",
                metric=name,
                value=value,
                threshold=thresholds['critical']
            )
        elif abs_value >= thresholds['warning']:
            self._create_alert(
                AlertLevel.WARNING,
                f"Warning: {name}",
                f"{name} = {value:.4f} exceeds warning threshold {thresholds['warning']}",
                metric=name,
                value=value,
                threshold=thresholds['warning']
            )
    
    def _create_alert(self, level: AlertLevel, title: str, message: str, **kwargs):
        """Cree et envoie une alerte."""
        alert = Alert(level=level, title=title, message=message, **kwargs)
        self.alerts.append(alert)
        
        # Simulate sending
        print(f"[{level.value.upper()}] {title}: {message}")
        
        if level == AlertLevel.CRITICAL:
            self._send_sms(alert)
            self._send_email(alert)
        elif level == AlertLevel.WARNING:
            self._send_email(alert)
    
    def _send_sms(self, alert: Alert):
        """Simule l'envoi SMS."""
        print(f"  -> SMS sent: {alert.title}")
    
    def _send_email(self, alert: Alert):
        """Simule l'envoi email."""
        print(f"  -> Email sent: {alert.title}")
    
    def get_dashboard_data(self) -> Dict:
        """Retourne les donnees pour le dashboard."""
        return {
            'metrics': {
                name: {
                    'current': values[-1][1] if values else None,
                    'history': [(t.isoformat(), v) for t, v in values[-100:]]
                }
                for name, values in self.metrics_history.items()
            },
            'alerts': [
                {
                    'level': a.level.value,
                    'title': a.title,
                    'message': a.message,
                    'timestamp': a.timestamp.isoformat()
                }
                for a in self.alerts[-50:]
            ],
            'alert_counts': {
                'critical': sum(1 for a in self.alerts if a.level == AlertLevel.CRITICAL),
                'warning': sum(1 for a in self.alerts if a.level == AlertLevel.WARNING),
                'info': sum(1 for a in self.alerts if a.level == AlertLevel.INFO)
            }
        }


# Demonstration
monitor = MonitoringSystem()

print("Monitoring System Demo")
print("="*50)

# Simuler des metriques
print("\nSimulating metrics...")
monitor.record_metric('drawdown', -0.03)  # OK
monitor.record_metric('drawdown', -0.06)  # Warning
monitor.record_metric('drawdown', -0.12)  # Critical

print(f"\nTotal alerts: {len(monitor.alerts)}")

Monitoring System Demo==================================================Simulating metrics...[WARNING] Warning: drawdown: drawdown = -0.0600 exceeds warning threshold 0.05  -> Email sent: Warning: drawdown[CRITICAL] Critical: drawdown: drawdown = -0.1200 exceeds critical threshold 0.1  -> SMS sent: Critical: drawdown  -> Email sent: Critical: drawdownTotal alerts: 2

Lecture — la graduation des alertes warning → critical. La démonstration alimente le monitoring avec trois valeurs de drawdown : -0,03 (aucune alerte, sous les deux seuils), -0,06 (franchit le seuil warning de 0,05 → Email), -0,12 (franchit le seuil critical de 0,10 → SMS + Email). La sortie confirme la hiérarchie : le warning ne déclenche qu’un email, le critical escalade vers le SMS — le canal le plus immédiat — en plus de l’email. Le compteur final Total alerts: 2 montre que chaque dépassement est indépendant : deux métriques franchissent un seuil, deux alertes sont créées. Cette architecture de seuils par métrique (drawdown, daily_loss, sharpe_deviation, slippage_bps, fill_time_seconds) est le cœur du déploiement : sans monitoring, une dérive de drawdown n’est détectée qu’au rapport suivant ; avec, l’opérateur reçoit l’alerte au moment où le seuil est franchi.

---

Exercice 3.1 : Orchestrateur multi-stratégies

Un système de production combine plusieurs stratégies. L’orchestrateur gere l’allocation inter-stratégies.

Objectif : Implementer un allocateur dynamique basé sur la performance recente.

Règles : - 3 stratégies simulees avec des rendements différents - Alloquer proportionnellement au Sharpe ratio glissant (20 jours) - Poids minimum 10% par stratégie, maximum 50% - Rebalance hebdomadaire - Affichez les allocations et la performance combinee

Indices : - # Indice : sharpe_glissant = mean(returns_20j) / max(std(returns_20j), 1e-6) - # Indice : Normalisez les Sharpe en poids puis clippez entre 0.10 et 0.50

# Exercice 3.1 : Allocateur multi-strategies
# TODO etudiant : Implementer un allocateur dynamique
# Indice : Sharpe glissant + normalisation + contraintes min/max
# Etape 1 : Simuler les rendements des 3 strategies
# Etape 2 : Calculer les Sharpe glissants
# Etape 3 : Convertir en allocations contraintes
# Etape 4 : Afficher allocations et performance combinee

result = None  # TODO etudiant : remplacer par l'allocateur
print("Exercice a completer")

---

Partie 5 : Multi-Strategy Orchestration (10 min)

Architecture Multi-Stratégies

flowchart TD
    O["ORCHESTRATOR"] --> SA["Strategy A (Momentum)"]
    O --> SB["Strategy B (Mean Rev)"]
    O --> SC["Strategy C (ML)"]
    SA --> RM["Risk Manager"]
    SB --> RM
    SC --> RM
    RM --> EX["Exécution"]

Allocation de Capital

Méthode	Description	Use Case
Fixe	33%/33%/33%	Simple, debut
Risk Parity	Selon volatilite	Equilibrer le risque
Performance	Selon Sharpe recent	Recompenser gagnants
Kelly	Optimal théorique	Agressif

Ancrage théorique — allocation de capital multi-stratégies. La ligne Risk Parity renvoie au risk budgeting où chaque stratégie reçoit un poids inversement proportionnel à sa contribution de risque, formalisé par Maillard, Roncalli & Teiletche (2010) The Properties of Equally Weighted Risk Contribution Portfolios, Journal of Portfolio Management 36(4):60–70 — le portefeuille de parité de risque équilibre les contributions de risque plutôt que les capitaux. La ligne Kelly renvoie au critère de Kelly (Kelly 1956 A New Interpretation of Information Rate, Bell System Technical Journal 35:917–926), qui maximise le taux de croissance attendu du log-capital et donne la fraction optimale à risquer par stratégie — théoriquement optimal sous hypothèses fortes (probabilités connues, fonction d’utilité logarithmique), d’où le qualificatif « optimal théorique » et l’usage prudent « agressif » en pratique.

# Orchestrateur multi-strategies

@dataclass
class StrategyConfig:
    """Configuration d'une strategie."""
    name: str
    project_id: str
    target_allocation: float  # 0 to 1
    max_allocation: float     # Upper bound
    min_allocation: float     # Lower bound
    is_active: bool = True
    
    # Performance tracking
    recent_sharpe: float = 0.0
    recent_return: float = 0.0
    current_allocation: float = 0.0


class MultiStrategyOrchestrator:
    """
    Orchestre plusieurs strategies de trading.
    """
    
    def __init__(self, total_capital: float = 1000000.0):
        self.total_capital = total_capital
        self.strategies: Dict[str, StrategyConfig] = {}
        self.allocation_method = "risk_parity"  # fixed, risk_parity, performance
    
    def add_strategy(self, config: StrategyConfig):
        """Ajoute une strategie."""
        self.strategies[config.name] = config
        print(f"Added strategy: {config.name} (target: {config.target_allocation:.0%})")
    
    def update_performance(self, name: str, sharpe: float, ret: float):
        """Met a jour les performances d'une strategie."""
        if name in self.strategies:
            self.strategies[name].recent_sharpe = sharpe
            self.strategies[name].recent_return = ret
    
    def compute_allocations(self) -> Dict[str, float]:
        """Calcule les allocations optimales."""
        active = [s for s in self.strategies.values() if s.is_active]
        
        if not active:
            return {}
        
        if self.allocation_method == "fixed":
            return self._fixed_allocation(active)
        elif self.allocation_method == "risk_parity":
            return self._risk_parity_allocation(active)
        elif self.allocation_method == "performance":
            return self._performance_allocation(active)
        
        return {}
    
    def _fixed_allocation(self, strategies: List[StrategyConfig]) -> Dict[str, float]:
        """Allocation fixe selon target."""
        total_target = sum(s.target_allocation for s in strategies)
        allocations = {}
        
        for s in strategies:
            alloc = s.target_allocation / total_target if total_target > 0 else 1/len(strategies)
            allocations[s.name] = np.clip(alloc, s.min_allocation, s.max_allocation)
        
        # Normalize
        total = sum(allocations.values())
        return {k: v/total for k, v in allocations.items()}
    
    def _risk_parity_allocation(self, strategies: List[StrategyConfig]) -> Dict[str, float]:
        """Allocation Risk Parity (inverse de la volatilite)."""
        # Utiliser 1/vol comme proxy (ici on utilise Sharpe comme proxy)
        weights = {}
        for s in strategies:
            # Eviter division par zero
            risk_contribution = 1 / (abs(s.recent_sharpe) + 0.1)
            weights[s.name] = risk_contribution
        
        total = sum(weights.values())
        allocations = {k: v/total for k, v in weights.items()}
        
        # Apply constraints
        for s in strategies:
            allocations[s.name] = np.clip(
                allocations[s.name],
                s.min_allocation,
                s.max_allocation
            )
        
        # Re-normalize
        total = sum(allocations.values())
        return {k: v/total for k, v in allocations.items()}
    
    def _performance_allocation(self, strategies: List[StrategyConfig]) -> Dict[str, float]:
        """Allocation basee sur la performance recente."""
        weights = {}
        for s in strategies:
            # Poids proportionnel au Sharpe positif
            weight = max(0, s.recent_sharpe)
            weights[s.name] = weight + 0.1  # Minimum weight
        
        total = sum(weights.values())
        allocations = {k: v/total for k, v in weights.items()}
        
        # Apply constraints
        for s in strategies:
            allocations[s.name] = np.clip(
                allocations[s.name],
                s.min_allocation,
                s.max_allocation
            )
        
        total = sum(allocations.values())
        return {k: v/total for k, v in allocations.items()}
    
    def rebalance(self) -> Dict[str, float]:
        """Execute le rebalancing."""
        allocations = self.compute_allocations()
        
        print("\nRebalancing allocations:")
        for name, alloc in allocations.items():
            capital = self.total_capital * alloc
            old_alloc = self.strategies[name].current_allocation
            change = alloc - old_alloc
            
            self.strategies[name].current_allocation = alloc
            print(f"  {name}: {alloc:.1%} (${capital:,.0f}) [{change:+.1%}]")
        
        return allocations


# Demonstration
orchestrator = MultiStrategyOrchestrator(total_capital=1_000_000)

# Ajouter strategies
strategies = [
    StrategyConfig("Momentum", "proj1", 0.4, 0.5, 0.2, recent_sharpe=1.8, recent_return=0.15),
    StrategyConfig("MeanReversion", "proj2", 0.3, 0.4, 0.15, recent_sharpe=1.2, recent_return=0.08),
    StrategyConfig("MLSignals", "proj3", 0.3, 0.4, 0.15, recent_sharpe=2.1, recent_return=0.20),
]

for s in strategies:
    orchestrator.add_strategy(s)

print("\n" + "="*50)
print("Testing allocation methods:")
print("="*50)

for method in ["fixed", "risk_parity", "performance"]:
    orchestrator.allocation_method = method
    print(f"\n{method.upper()} allocation:")
    allocs = orchestrator.compute_allocations()
    for name, alloc in allocs.items():
        print(f"  {name}: {alloc:.1%}")

Added strategy: Momentum (target: 40%)Added strategy: MeanReversion (target: 30%)Added strategy: MLSignals (target: 30%)==================================================Testing allocation methods:==================================================FIXED allocation:  Momentum: 40.0%  MeanReversion: 30.0%  MLSignals: 30.0%RISK_PARITY allocation:  Momentum: 31.3%  MeanReversion: 41.6%  MLSignals: 27.0%PERFORMANCE allocation:  Momentum: 35.4%  MeanReversion: 24.3%  MLSignals: 40.3%

Lecture — trois méthodes d’allocation, trois philosophies de risque. Sur les mêmes stratégies (Momentum, MeanReversion, MLSignals), l’orchestrateur produit : FIXED 40/30/30 (les poids cibles définis à l’ajout, aucun calcul) ; RISK_PARITY 31,3/41,6/27,0 — la stratégie la plus stable, MeanReversion, reçoit le poids le plus fort, car la parité de risque pondère par l’inverse de la volatilité (ici approchée par le Sharpe) : on récompense la stabilité, pas le rendement ; PERFORMANCE 35,4/24,3/40,3 — MLSignals, la stratégie au meilleur Sharpe récent (2,1), prend la tête, la pondération suit les performances passées. Le contraste entre les deux méthodes est le vrai sujet : la parité de risque équilibre les contributions de risque (une stratégie volatile ne peut pas dominer le portefeuille), la pondération performance concentre le capital sur les gagnants récents et accepte donc une volatilité plus élevée. L’exercice 3.1 demande d’implémenter ce calcul avec contraintes min/max ; le 3.2 ajoute une contrainte de corrélation.

Exercice 3.2 : Allocation avec contrainte de correlation

Ajoutez une contrainte de correlation a l’orchestrateur multi-stratégies : si deux stratégies ont une correlation > 0.7, reduisez l’allocation de la moins performante.

Indices : - # Indice : Calculez la correlation entre les returns des stratégies - # Étape 1 : Simuler les returns de 3 stratégies - # Étape 2 : Calculer la matrice de correlation - # Étape 3 : Reduire l’allocation des stratégies correllees

# Exercice 3.2 : Allocation avec contrainte de correlation
# TODO etudiant : Reduire l'allocation des strategies trop correllees (>0.7)
# Etape 1 : Simuler les returns de 3 strategies
# Etape 2 : Calculer la matrice de correlation
# Etape 3 : Reduire l'allocation si corr > 0.7
adjusted_allocations = None  # TODO etudiant : remplacer par les allocations
print("Exercice a completer : Allocation avec contrainte de correlation")

Exercice a completer

---

Partie 6 : Checklist et Best Practices (10 min)

Pre-Deployment Checklist

# Checklist de deploiement

@dataclass
class ChecklistItem:
    """Item de checklist."""
    category: str
    description: str
    is_critical: bool
    is_completed: bool = False
    notes: str = ""


class DeploymentChecklist:
    """
    Checklist de deploiement pour algorithmes de trading.
    """
    
    def __init__(self):
        self.items: List[ChecklistItem] = self._create_default_checklist()
    
    def _create_default_checklist(self) -> List[ChecklistItem]:
        return [
            # Backtest Validation
            ChecklistItem("Backtest", "Minimum 3 ans de backtest", True),
            ChecklistItem("Backtest", "Sharpe Ratio > 1.0", True),
            ChecklistItem("Backtest", "Max Drawdown < 20%", True),
            ChecklistItem("Backtest", "Win rate et profit factor documentes", False),
            ChecklistItem("Backtest", "Test sur differentes periodes (rolling)", False),
            
            # Paper Trading
            ChecklistItem("Paper", "Minimum 2 semaines en paper trading", True),
            ChecklistItem("Paper", "Performance proche du backtest (+/- 30%)", True),
            ChecklistItem("Paper", "Slippage mesure et acceptable", True),
            ChecklistItem("Paper", "Fill times documentes", False),
            ChecklistItem("Paper", "Aucune erreur non geree", True),
            
            # Risk Management
            ChecklistItem("Risk", "Max position size defini (< 20%)", True),
            ChecklistItem("Risk", "Max daily loss defini (< 2%)", True),
            ChecklistItem("Risk", "Max total drawdown defini (< 10%)", True),
            ChecklistItem("Risk", "Stop-loss implementes", True),
            ChecklistItem("Risk", "Auto-liquidation sur breach", True),
            
            # Monitoring
            ChecklistItem("Monitoring", "Alertes SMS/Email configurees", True),
            ChecklistItem("Monitoring", "Daily reports actives", False),
            ChecklistItem("Monitoring", "Dashboard accessible", False),
            ChecklistItem("Monitoring", "Contact d'urgence defini", True),
            
            # Infrastructure
            ChecklistItem("Infra", "Broker credentials verifies", True),
            ChecklistItem("Infra", "Capital broker correct", True),
            ChecklistItem("Infra", "API keys en secrets (pas dans code)", True),
            ChecklistItem("Infra", "Backup strategy definie", False),
            
            # Documentation
            ChecklistItem("Docs", "Strategy document a jour", False),
            ChecklistItem("Docs", "Risk limits documentes", True),
            ChecklistItem("Docs", "Procedure de rollback definie", True),
            ChecklistItem("Docs", "Contacts et escalation documentes", False),
        ]
    
    def check(self, category: str, description: str, completed: bool = True, notes: str = ""):
        """Marque un item comme complete ou non."""
        for item in self.items:
            if item.category == category and description in item.description:
                item.is_completed = completed
                item.notes = notes
                return
    
    def is_ready(self) -> Tuple[bool, List[str]]:
        """Verifie si tous les items critiques sont completes."""
        missing = [
            f"[{item.category}] {item.description}"
            for item in self.items
            if item.is_critical and not item.is_completed
        ]
        return len(missing) == 0, missing
    
    def report(self) -> str:
        """Genere un rapport de la checklist."""
        ready, missing = self.is_ready()
        
        # Group by category
        categories = {}
        for item in self.items:
            if item.category not in categories:
                categories[item.category] = []
            categories[item.category].append(item)
        
        report = f"""
{'='*60}
DEPLOYMENT CHECKLIST
{'='*60}
Status: {'READY' if ready else 'NOT READY'}
"""
        
        for cat, items in categories.items():
            report += f"\n{cat.upper()}:\n"
            for item in items:
                status = "[X]" if item.is_completed else "[ ]"
                critical = "*" if item.is_critical else " "
                report += f"  {status}{critical} {item.description}\n"
                if item.notes:
                    report += f"        Note: {item.notes}\n"
        
        if missing:
            report += f"\n{'='*60}\n"
            report += "CRITICAL ITEMS MISSING:\n"
            for m in missing:
                report += f"  - {m}\n"
        
        report += f"\n* = Critical item\n"
        
        return report


# Demonstration
checklist = DeploymentChecklist()

# Simuler completion de certains items
checklist.check("Backtest", "Minimum 3 ans", True)
checklist.check("Backtest", "Sharpe Ratio", True)
checklist.check("Backtest", "Max Drawdown", True)
checklist.check("Paper", "Minimum 2 semaines", True)
checklist.check("Paper", "Performance proche", True, "Within 25% of backtest")
checklist.check("Risk", "Max position size", True)
checklist.check("Risk", "Max daily loss", True)
checklist.check("Monitoring", "Alertes SMS", True)
checklist.check("Infra", "Broker credentials", True)
checklist.check("Infra", "API keys", True)

print(checklist.report())

============================================================DEPLOYMENT CHECKLIST============================================================Status: NOT READYBACKTEST:  [X]* Minimum 3 ans de backtest  [X]* Sharpe Ratio > 1.0  [X]* Max Drawdown < 20%  [ ]  Win rate et profit factor documentes  [ ]  Test sur differentes periodes (rolling)PAPER:  [X]* Minimum 2 semaines en paper trading  [X]* Performance proche du backtest (+/- 30%)        Note: Within 25% of backtest  [ ]* Slippage mesure et acceptable  [ ]  Fill times documentes  [ ]* Aucune erreur non gereeRISK:  [X]* Max position size defini (< 20%)  [X]* Max daily loss defini (< 2%)  [ ]* Max total drawdown defini (< 10%)  [ ]* Stop-loss implementes  [ ]* Auto-liquidation sur breachMONITORING:  [X]* Alertes SMS/Email configurees  [ ]  Daily reports actives  [ ]  Dashboard accessible  [ ]* Contact d'urgence definiINFRA:  [X]* Broker credentials verifies  [ ]* Capital broker correct  [X]* API keys en secrets (pas dans code)  [ ]  Backup strategy definieDOCS:  [ ]  Strategy document a jour  [ ]* Risk limits documentes  [ ]* Procedure de rollback definie  [ ]  Contacts et escalation documentes============================================================CRITICAL ITEMS MISSING:  - [Paper] Slippage mesure et acceptable  - [Paper] Aucune erreur non geree  - [Risk] Max total drawdown defini (< 10%)  - [Risk] Stop-loss implementes  - [Risk] Auto-liquidation sur breach  - [Monitoring] Contact d'urgence defini  - [Infra] Capital broker correct  - [Docs] Risk limits documentes  - [Docs] Procedure de rollback definie* = Critical item

Lecture — la checklist est un état de préparation, pas un formulaire. Le rapport final affiche NOT READY avec 9 items critiques manquants : côté Paper, le slippage mesuré et la gestion des erreurs ; côté Risk, le drawdown maximal total, les stop-loss et l’auto-liquidation ; côté Monitoring, le contact d’urgence ; côté Infra, le capital broker correct ; côté Docs, les limites de risque et la procédure de rollback. Les cases cochées montrent le chemin fait : backtest validé (3 ans, Sharpe > 1,0, drawdown < 20 %), paper avec performance within 25 % du backtest, position et perte journalière définies, alertes configurées, secrets hors code. La leçon : chaque item critique non coché est un risque accepté consciemment — et la méthode is_ready() force cette conscience en refusant un verdict READY tant qu’un seul item critique manque. L’auto-évaluation honnête NOT READY vaut mieux qu’un passage en live prématuré.

On met en place les mécanismes de surveillance et d’alertes pour superviser l’algorithme déployé. En production, la stratégie vit sans son auteur : ce qui remplace la vigilance humaine, c’est un système de monitoring gradué — des métriques continues (exposition, turnover, erreur de tracking vs benchmark), des seuils d’alerte calibrés en deux niveaux (warning = investigation, critical = intervention immédiate), et un journal d’événements qui permet de reconstituer a posteriori pourquoi le système a pris une décision. La règle pratique : toute alerte doit être actionnable — une métrique surveillée sans seuil ni conséquence est de la décor, pas du contrôle.

La démonstration ci-dessous exécute un cycle de monitoring complet sur des journées simulées : elle montre la graduation warning → critical quand les conditions se dégradent, et le retour au vert quand elles se rétablissent.

# Resume et meilleures pratiques

print("="*70)
print("RESUME : PRODUCTION DEPLOYMENT")
print("="*70)

best_practices = """
1. CYCLE DE VIE
   - Draft -> Testing -> Paper -> Live
   - Criteres stricts entre chaque etape
   - Documentation a chaque transition

2. PAPER TRADING
   - Minimum 2 semaines avant live
   - Comparer avec backtest expectations
   - Tracker slippage et fill times
   - Valider la gestion des erreurs

3. RISK MANAGEMENT
   - Max position size: 20%
   - Max daily loss: 2%
   - Max total drawdown: 10%
   - Auto-liquidation sur breach

4. MONITORING
   - Alertes SMS pour critical
   - Email pour warnings
   - Daily reports automatiques
   - Dashboard accessible 24/7

5. MULTI-STRATEGY
   - Diversifier les approches
   - Risk parity allocation
   - Rebalancing mensuel
   - Correlation monitoring

6. DOCUMENTATION
   - Strategy document complet
   - Risk limits clairs
   - Procedure de rollback
   - Contacts d'urgence
"""

print(best_practices)

======================================================================RESUME : PRODUCTION DEPLOYMENT======================================================================1. CYCLE DE VIE   - Draft -> Testing -> Paper -> Live   - Criteres stricts entre chaque etape   - Documentation a chaque transition2. PAPER TRADING   - Minimum 2 semaines avant live   - Comparer avec backtest expectations   - Tracker slippage et fill times   - Valider la gestion des erreurs3. RISK MANAGEMENT   - Max position size: 20%   - Max daily loss: 2%   - Max total drawdown: 10%   - Auto-liquidation sur breach4. MONITORING   - Alertes SMS pour critical   - Email pour warnings   - Daily reports automatiques   - Dashboard accessible 24/75. MULTI-STRATEGY   - Diversifier les approches   - Risk parity allocation   - Rebalancing mensuel   - Correlation monitoring6. DOCUMENTATION   - Strategy document complet   - Risk limits clairs   - Procedure de rollback   - Contacts d'urgence

Lecture — la synthèse condense la discipline en six blocs. Le résumé final restitue l’ensemble du notebook : (1) le cycle de vie draft -> testing -> paper -> live avec critères stricts à chaque transition et documentation systématique ; (2) le paper trading — 2 semaines minimum, comparaison aux attentes du backtest, suivi du slippage et des temps de remplissage, validation de la gestion des erreurs ; (3) le risk management chiffré — position max 20 %, perte journalière max 2 %, drawdown total max 10 %, auto-liquidation en cas de dépassement ; (4) le monitoring — SMS pour les alertes critiques, email pour les warnings, rapports quotidiens automatiques, dashboard accessible ; (5) le multi-strategy — diversification, allocation risk parity, rebalancement mensuel, surveillance des corrélations ; (6) la documentation — stratégie, limites de risque, procédure de rollback, contacts d’urgence. Ces six blocs sont exactement les sections du notebook : le résumé fonctionne comme la page de garde de votre discipline de déploiement — à relire avant chaque passage en live.

---

Conclusion : Serie QuantConnect Complete

Recapitulatif de la Serie (27 Notebooks)

Phase	Notebooks	Contenu
1. Fondations	01-04	Setup, Platform, Data, Research
2. Assets	05-08	Universe, Options, Futures, Multi-Asset
3. Trading	09-12	Orders, Risk, Indicators, Backtesting
4. Framework	13-15	Alpha, Portfolio, Optimization
5. Data	16-18	Alternative, Sentiment, Features
6. ML	19-21	Classification, Regression, Portfolio
7. DL SOTA	22-24	Time Series, Mamba/SSMs, VAE/HMM
8. Production	25-27	RL, LLMs, Deployment

Parcours Recommande

Debutant       Intermediaire       Avance          Expert
01-04 -------> 05-12 -----------> 13-21 -------> 22-27
(1 semaine)    (2 semaines)       (3 semaines)   (2 semaines)

Ressources Complementaires

• QuantConnect Documentation
• QuantConnect Forum
• LEAN Engine GitHub
• Hands-On AI Trading Book

Références

• Kelly, J.L. Jr. (1956). A New Interpretation of Information Rate. Bell System Technical Journal, 35, 917–926. doi:10.1002/j.1538-7305.1956.tb03809.x
• Maillard, S., Roncalli, T. & Teiletche, J. (2010). The Properties of Equally Weighted Risk Contribution Portfolios. The Journal of Portfolio Management, 36(4), 60–70. doi:10.3905/jpm.2010.36.4.060

---

Felicitations ! Vous avez complete la serie QuantConnect AI Trading.

Bonne chance dans vos stratégies de trading algorithmique !