From 0209c61980c7db2844466e720914e9ede34a1640 Mon Sep 17 00:00:00 2001 From: sora Date: Tue, 7 Jul 2026 08:28:01 +0000 Subject: [PATCH 01/17] fix: Add mlflow token & uri --- app.py | 24 +++++++++++++----------- 1 file changed, 13 insertions(+), 11 deletions(-) diff --git a/app.py b/app.py index f781ead..8443854 100644 --- a/app.py +++ b/app.py @@ -18,7 +18,7 @@ # ============================================================================= -# 1. CONFIGURATION & STYLE +# CONFIGURATION & STYLE # ============================================================================= st.set_page_config( page_title="AlphaEdge Dashboard", @@ -30,8 +30,6 @@ BASE_DIR = Path(__file__).resolve().parent HF_REPO_ID = os.getenv("HF_REPO_ID", "soradata/alphaedge-data") - -# --- MLflow : mêmes réglages que train.py --- MLFLOW_TRACKING_URI = "https://soradata-alphaedge-registry.hf.space" HF_TOKEN = os.getenv("HF_TOKEN") MLFLOW_ENABLED = bool(HF_TOKEN) @@ -40,11 +38,7 @@ os.environ["MLFLOW_TRACKING_USERNAME"] = "SORADATA" os.environ["MLFLOW_TRACKING_PASSWORD"] = HF_TOKEN mlflow.set_tracking_uri(MLFLOW_TRACKING_URI) - -# Où train.py écrit les model_card.json locaux (MODEL_DIR/{market}/model_card.json). -# Ajuste ce chemin si ton const.MODEL_DIR pointe ailleurs sur ce déploiement. MODEL_DIR = BASE_DIR / "models" - MARKET_OPTIONS = ["CAC40", "BRVM"] st.markdown(""" @@ -94,7 +88,7 @@ # ============================================================================= -# 2. CHARGEMENT DES DONNÉES DEPUIS HUGGING FACE (par marché) +# CHARGEMENT DES DONNÉES DEPUIS HUGGING FACE (par marché) # ============================================================================= @st.cache_data(ttl=600, show_spinner=False) @@ -138,7 +132,7 @@ def load_all_data(market: str): # ============================================================================= -# 3. FONCTIONS UTILITAIRES — KPIs / MARCHÉ +# FONCTIONS UTILITAIRES — KPIs / MARCHÉ # ============================================================================= def display_kpi_card(label, value, is_percent=True, color_code=False, prefix="", minimal=False): @@ -310,7 +304,15 @@ def get_champion_metrics(market: str): if not df_hist.empty: last_dt = df_hist.index[-1] days_old = (datetime.now() - last_dt).days - status_icon, status_text = ("🟢", "● System Online") if days_old == 0 else ("🟡", "○ Data Slightly Old") if days_old <= 3 else ("🔴", "○ Data Outdated") + + # MODIFICATION ICI : Considérer les données de la veille comme "System Online" (Stratégie D-1) + if days_old <= 1: + status_icon, status_text = "🟢", "● System Online" + elif days_old <= 3: + status_icon, status_text = "🟡", "○ Data Slightly Old" + else: + status_icon, status_text = "🔴", "○ Data Outdated" + st.sidebar.info(f"Last Update: {last_dt.date()}") st.sidebar.markdown(f"{status_icon} {status_text}") else: @@ -657,4 +659,4 @@ def get_champion_metrics(market: str):

Risques : Tout investissement comporte des risques. Les performances passées ne garantissent pas les résultats futurs.

Responsabilité : Consultez un conseiller financier agréé avant toute décision d'investissement.

-""", unsafe_allow_html=True) +""", unsafe_allow_html=True) \ No newline at end of file From 9cbff6f1a248c5ef3c9c2bee4efce9fc46d112a0 Mon Sep 17 00:00:00 2001 From: sora Date: Tue, 7 Jul 2026 08:53:18 +0000 Subject: [PATCH 02/17] fix: Add .xs rather than bool & vectorize simulation --- app.py | 7 +- src/pipeline/backtest.py | 570 +++++++++++++++++++++++++++++++-------- 2 files changed, 455 insertions(+), 122 deletions(-) diff --git a/app.py b/app.py index 8443854..9ec9e3f 100644 --- a/app.py +++ b/app.py @@ -304,15 +304,14 @@ def get_champion_metrics(market: str): if not df_hist.empty: last_dt = df_hist.index[-1] days_old = (datetime.now() - last_dt).days - - # MODIFICATION ICI : Considérer les données de la veille comme "System Online" (Stratégie D-1) + + # Considérer les données de la veille comme "System Online" (Stratégie D-1) if days_old <= 1: status_icon, status_text = "🟢", "● System Online" elif days_old <= 3: status_icon, status_text = "🟡", "○ Data Slightly Old" else: status_icon, status_text = "🔴", "○ Data Outdated" - st.sidebar.info(f"Last Update: {last_dt.date()}") st.sidebar.markdown(f"{status_icon} {status_text}") else: @@ -322,7 +321,7 @@ def get_champion_metrics(market: str): ticker_val_path = BASE_DIR / "config" / selected_market / "ticker_validation.json" if not ticker_val_path.exists(): - ticker_val_path = BASE_DIR / "ticker_validation.json" # fallback ancien chemin unique + ticker_val_path = BASE_DIR / "ticker_validation.json" if ticker_val_path.exists(): with st.sidebar.expander("🔍 Ticker Health", expanded=False): try: diff --git a/src/pipeline/backtest.py b/src/pipeline/backtest.py index 2c6e67a..72a3035 100644 --- a/src/pipeline/backtest.py +++ b/src/pipeline/backtest.py @@ -1,4 +1,33 @@ +""" +backtest.py +=========== + +Moteur de backtest et de génération de signaux live pour AlphaEdge. + +Architecture du module +----------------------- +1. Construction de portefeuille : sélection des tickers + optimisation Markowitz +2. Moteur de simulation : vectorisé (numpy), pas de boucle Python par jour +3. Analytics de performance : Sharpe, Sortino, Calmar, Alpha/Beta, Information Ratio, + Tracking Error, coût de turnover +4. API publique : backtest_strategy_with_rebalancing / generate_live_signals + +Notes de conception +-------------------- +- Les positions non allouées (somme des poids < 1) sont conservées en cash à + rendement nul, explicitement comptabilisées — pas de "cash fantôme". +- Les coûts de transaction sont appliqués à chaque rebalancement, proportionnellement + au turnover réel entre l'allocation cible et l'allocation *drifted* (dérivée + des variations de prix depuis le dernier rebalancement). +- Aucune métrique de performance passée ne constitue une garantie de résultats + futurs. Ce module sert à l'analyse quantitative, pas à du conseil en investissement. +""" + +from __future__ import annotations + +import logging from typing import Any, Dict, List, Optional, Tuple + import numpy as np import pandas as pd from pypfopt import EfficientFrontier, risk_models, expected_returns, objective_functions @@ -10,7 +39,7 @@ MIN_STOCKS_OPTIM, MAX_STOCKS_SELECT, PROBA_MIN, - WEIGHT_BOUNDS + WEIGHT_BOUNDS, ) from src.features.alpha_features import add_all_features from src.utils.logger import setup_logger @@ -18,94 +47,217 @@ logger = setup_logger("backtest") +# Nombre de jours de forward-fill toléré pour un prix manquant avant de +# considérer le titre comme illiquide/délisté et de l'exclure de la fenêtre. +MAX_PRICE_FFILL_DAYS = 5 + +# Seuil de features manquantes (moyenne sur toutes les colonnes) au-delà +# duquel on log un warning explicite plutôt que d'imputer silencieusement. +MAX_ACCEPTABLE_MISSING_RATIO = 0.05 + + +# ============================================================================= +# 1. CONSTRUCTION DE PORTEFEUILLE +# ============================================================================= + +def get_optimal_weights( + prices_df: pd.DataFrame, + risk_free_rate: float = RISK_FREE_RATE, +) -> Tuple[Dict[str, float], str]: + """ + Optimise les poids du portefeuille (Max Sharpe -> Min Vol -> Equal Weight en cascade). -def get_optimal_weights(prices_df: pd.DataFrame, risk_free_rate: float = RISK_FREE_RATE) -> Tuple[Dict[str, float], str]: - if prices_df.shape[1] < MIN_STOCKS_OPTIM: - n = prices_df.shape[1] - return {t: 1.0 / n for t in prices_df.columns}, "equal_weight" + Args: + prices_df: Prix ajustés, colonnes = tickers, index = dates. + risk_free_rate: Taux sans risque annualisé utilisé pour Max Sharpe. + + Returns: + (weights, method) où method indique la stratégie effectivement utilisée, + utile pour le monitoring de la robustesse de l'optimiseur en production. + """ + n_assets = prices_df.shape[1] + if n_assets < MIN_STOCKS_OPTIM: + return {t: 1.0 / n_assets for t in prices_df.columns}, "equal_weight" try: mu = expected_returns.ema_historical_return(prices_df, frequency=TRADING_DAYS_YEAR, span=252) - S = risk_models.CovarianceShrinkage(prices_df, frequency=TRADING_DAYS_YEAR).ledoit_wolf() + cov = risk_models.CovarianceShrinkage(prices_df, frequency=TRADING_DAYS_YEAR).ledoit_wolf() - ef = EfficientFrontier(mu, S, weight_bounds=WEIGHT_BOUNDS) + ef = EfficientFrontier(mu, cov, weight_bounds=WEIGHT_BOUNDS) ef.add_objective(objective_functions.L2_reg, gamma=0.1) ef.max_sharpe(risk_free_rate=risk_free_rate) - return dict(ef.clean_weights()), "max_sharpe" + weights = dict(ef.clean_weights()) + _log_concentration(weights) + return weights, "max_sharpe" - except Exception as e1: - logger.warning(f"Max Sharpe failed ({e1}) -> min_volatility fallback.") + except Exception as exc: + logger.warning(f"Max Sharpe a échoué ({exc}) -> fallback min_volatility.") try: mu = expected_returns.ema_historical_return(prices_df, frequency=TRADING_DAYS_YEAR, span=252) - S = risk_models.CovarianceShrinkage(prices_df, frequency=TRADING_DAYS_YEAR).ledoit_wolf() + cov = risk_models.CovarianceShrinkage(prices_df, frequency=TRADING_DAYS_YEAR).ledoit_wolf() + + ef_minvol = EfficientFrontier(mu, cov, weight_bounds=WEIGHT_BOUNDS) + ef_minvol.min_volatility() + + weights = dict(ef_minvol.clean_weights()) + _log_concentration(weights) + return weights, "min_vol" - ef2 = EfficientFrontier(mu, S, weight_bounds=WEIGHT_BOUNDS) - ef2.min_volatility() + except Exception as exc2: + logger.warning(f"Min Vol a également échoué ({exc2}) -> fallback equal_weight.") + return {t: 1.0 / n_assets for t in prices_df.columns}, "equal_weight" - return dict(ef2.clean_weights()), "min_vol" - except Exception as e2: - logger.warning(f"Min vol failed ({e2}) -> equal_weight fallback.") - n = prices_df.shape[1] - return {t: 1.0 / n for t in prices_df.columns}, "equal_weight" +def _log_concentration(weights: Dict[str, float]) -> None: + """Log l'indice de Herfindahl-Hirschman du portefeuille (mesure de concentration).""" + active = np.array([w for w in weights.values() if w > 1e-6]) + if active.size == 0: + return + hhi = float(np.sum(active ** 2)) + logger.debug(f"Concentration du portefeuille (HHI) : {hhi:.3f} sur {active.size} positions.") -def _score_with_model(model: Any, X: pd.DataFrame) -> np.ndarray: - """Scoring robuste avec reindex pour gérer les lags macro manquants le jour J.""" +def _score_with_model(model: Any, features: pd.DataFrame) -> np.ndarray: + """ + Scoring robuste du modèle avec reindex pour gérer les colonnes manquantes + (ex : lags macro non disponibles le jour J). + + Toute imputation par zéro est loggée si elle dépasse un seuil raisonnable, + afin de détecter une dérive de la qualité de données en production plutôt + que de la masquer silencieusement. + """ if hasattr(model, "predict_proba"): - if hasattr(model, "features_"): - X_input = X.reindex(columns=model.features_).copy() - else: - X_input = X.copy() - X_input = X_input.fillna(0) - return model.predict_proba(X_input)[:, 1] + expected_cols = getattr(model, "features_", None) + x_input = features.reindex(columns=expected_cols).copy() if expected_cols else features.copy() + _warn_if_missing(x_input) + return model.predict_proba(x_input.fillna(0))[:, 1] if hasattr(model, "predict"): + expected_cols = None try: input_schema = model.metadata.get_input_schema() expected_cols = [c.name for c in input_schema.inputs] if input_schema else None except Exception: - expected_cols = None + pass - if expected_cols: - X_input = X.reindex(columns=expected_cols).copy() - else: - X_input = X.copy() - X_input = X_input.fillna(0) - preds = model.predict(X_input) + x_input = features.reindex(columns=expected_cols).copy() if expected_cols else features.copy() + _warn_if_missing(x_input) + preds = model.predict(x_input.fillna(0)) return np.asarray(preds).ravel() - raise TypeError(f"Type de modèle non supporté : {type(model)}") + raise TypeError(f"Type de modèle non supporté pour le scoring : {type(model)}") + + +def _warn_if_missing(x_input: pd.DataFrame) -> None: + if x_input.empty: + return + missing_ratio = float(x_input.isna().mean().mean()) + if missing_ratio > MAX_ACCEPTABLE_MISSING_RATIO: + logger.warning( + f"Scoring : {missing_ratio:.1%} de valeurs manquantes en moyenne dans les " + "features avant imputation à 0 — vérifier la fraîcheur des données amont." + ) def _generate_monthly_signals(month_data: pd.DataFrame, model: Any) -> pd.DataFrame: + """Score un snapshot (un ticker par ligne) et ajoute la colonne 'proba_upside'.""" if month_data.empty: return pd.DataFrame() try: - month_data = month_data.copy() - month_data["proba_upside"] = _score_with_model(model, month_data) - except Exception as e: - logger.error(f"Scoring échoué : {e}", exc_info=True) + scored = month_data.copy() + scored["proba_upside"] = _score_with_model(model, scored) + return scored + except Exception: + logger.error("Scoring du snapshot échoué.", exc_info=True) return pd.DataFrame() - return month_data -def _select_tickers(month_data: pd.DataFrame, proba_min: float = PROBA_MIN, max_stocks: int = MAX_STOCKS_SELECT) -> List[str]: +def _select_tickers( + month_data: pd.DataFrame, + proba_min: float = PROBA_MIN, + max_stocks: int = MAX_STOCKS_SELECT, +) -> List[str]: + """ + Sélectionne les tickers dont la probabilité de hausse dépasse le seuil, + triés par conviction décroissante, plafonnés à max_stocks. + + IMPORTANT : `month_data.index` doit être un index simple de tickers + (pas un MultiIndex ticker/date), sous peine de retourner des tuples + qui ne matcheront jamais les colonnes d'un DataFrame de prix en aval. + """ if "proba_upside" not in month_data.columns: return [] + if isinstance(month_data.index, pd.MultiIndex): + raise ValueError( + "_select_tickers attend un index simple de tickers ; reçu un MultiIndex. " + "Utiliser .xs(date, level='date') en amont pour aplatir l'index." + ) selected = month_data[month_data["proba_upside"] >= proba_min] if selected.empty: return [] - return selected.sort_values(by="proba_upside", ascending=False).head(max_stocks).index.tolist() + return selected.sort_values("proba_upside", ascending=False).head(max_stocks).index.tolist() + + +def _blend_with_conviction( + weights: Dict[str, float], + proba_by_ticker: pd.Series, + conviction_tilt: float = 0.0, +) -> Dict[str, float]: + """ + Incline les poids risk-based (Markowitz) vers les tickers à plus forte + conviction (proba de hausse plus élevée), dans l'esprit d'un overlay + "signal-weighted" utilisé par de nombreux quant funds. + + conviction_tilt = 0.0 -> comportement inchangé (poids Markowitz purs). + conviction_tilt = 1.0 -> poids entièrement proportionnels à la conviction. + """ + if conviction_tilt <= 0 or not weights: + return weights + + proba_aligned = proba_by_ticker.reindex(weights.keys()) + if proba_aligned.isna().any(): + proba_aligned = proba_aligned.fillna(proba_aligned.mean()) + if proba_aligned.sum() <= 0: + return weights + + conviction_weights = (proba_aligned / proba_aligned.sum()).to_dict() + blended = { + t: (1 - conviction_tilt) * weights[t] + conviction_tilt * conviction_weights[t] + for t in weights + } + total = sum(blended.values()) + return {t: w / total * sum(weights.values()) for t, w in blended.items()} if total > 0 else weights def _compute_turnover(new_alloc: Dict[str, float], old_alloc: Dict[str, float]) -> float: + """Turnover one-way = demi-somme des variations absolues de poids.""" all_tickers = set(new_alloc) | set(old_alloc) return sum(abs(new_alloc.get(t, 0.0) - old_alloc.get(t, 0.0)) for t in all_tickers) / 2.0 -def _simulate_daily_returns( +def _build_price_matrix(df_daily: pd.DataFrame, ffill_limit: int = MAX_PRICE_FFILL_DAYS) -> pd.DataFrame: + """ + Construit la matrice de prix ticker x date, avec un forward-fill *borné*. + + Un ffill illimité masquerait indéfiniment un titre délisté ou suspendu + (prix figé recopié à l'infini) ; on limite la tolérance à `ffill_limit` + jours puis on laisse les NaN résiduels, qui seront traités comme + "non éligible" plutôt que silencieusement valorisés à un prix obsolète. + """ + prices = df_daily["adj close"].unstack() + prices = prices.ffill(limit=ffill_limit) + stale_cols = prices.columns[prices.iloc[-1].isna()] + if len(stale_cols) > 0: + logger.debug(f"{len(stale_cols)} tickers exclus (prix obsolètes/manquants) : {list(stale_cols)[:10]}...") + return prices + + +# ============================================================================= +# 2. MOTEUR DE SIMULATION (VECTORISÉ) +# ============================================================================= + +def _simulate_period( allocation: Dict[str, float], drifted_allocation: Dict[str, float], trading_days: pd.DatetimeIndex, @@ -113,114 +265,288 @@ def _simulate_daily_returns( benchmark_returns: pd.Series, portfolio_value: float, benchmark_value: float, -) -> Tuple[list, float, float, Dict[str, float]]: - records = [] - is_first_day = True - stock_values = {t: portfolio_value * w for t, w in allocation.items()} if allocation else {} - - for date in trading_days: - bench_ret = benchmark_returns.get(date, 0.0) - if is_first_day: - turnover = _compute_turnover(allocation, drifted_allocation) - transaction_fees = portfolio_value * turnover * TRANSACTION_COST - portfolio_value -= transaction_fees - stock_values = {t: portfolio_value * w for t, w in allocation.items()} - is_first_day = False - - daily_portfolio_pnl = 0.0 - if stock_values: - for t in list(stock_values.keys()): - ret = daily_returns.loc[date, t] if (t in daily_returns.columns and date in daily_returns.index) else 0.0 - pnl = stock_values[t] * ret - stock_values[t] += pnl - daily_portfolio_pnl += pnl - - portfolio_value += daily_portfolio_pnl - benchmark_value *= (1 + bench_ret) - records.append({"Date": date, "Strategy": portfolio_value, "Benchmark": benchmark_value, "N_Stocks": len(stock_values)}) - - new_drifted_allocation = {} - if portfolio_value > 0 and stock_values: - new_drifted_allocation = {t: val / portfolio_value for t, val in stock_values.items()} - - return records, portfolio_value, benchmark_value, new_drifted_allocation - - -def compute_performance_metrics(results_df: pd.DataFrame, rebalance_log: pd.DataFrame, risk_free_rate: float = RISK_FREE_RATE) -> Dict[str, float]: + transaction_cost: float = TRANSACTION_COST, +) -> Tuple[pd.DataFrame, float, float, Dict[str, float]]: + """ + Simule l'évolution quotidienne du portefeuille sur une période de rebalancement, + de façon entièrement vectorisée (pas de boucle Python jour par jour). + + Le cash non investi (1 - somme des poids) est conservé à rendement nul et + explicitement réintégré dans la valeur totale du portefeuille — condition + nécessaire pour que le sizing (vol targeting, conviction tilt, etc.) soit + comptabilisé correctement. + + Returns: + period_df: colonnes Strategy / Benchmark / N_Stocks, indexées par date. + final_portfolio_value, final_benchmark_value: valeurs en fin de période. + new_drifted_allocation: poids réels (post-dérive de prix) en fin de période, + utilisés comme référence de turnover au prochain rebalancement. + """ + if len(trading_days) == 0: + empty = pd.DataFrame(columns=["Strategy", "Benchmark", "N_Stocks"]) + return empty, portfolio_value, benchmark_value, drifted_allocation + + turnover = _compute_turnover(allocation, drifted_allocation) + transaction_fees = portfolio_value * turnover * transaction_cost + portfolio_value -= transaction_fees + + tickers = list(allocation.keys()) + gross_exposure = sum(allocation.values()) if tickers else 0.0 + cash_amount = portfolio_value * (1.0 - gross_exposure) + + if tickers: + weights = np.array([allocation[t] for t in tickers]) + rets = daily_returns.reindex(index=trading_days, columns=tickers).fillna(0.0).to_numpy() + growth = np.cumprod(1.0 + rets, axis=0) # (n_days, n_tickers) + stock_values = portfolio_value * weights[np.newaxis, :] * growth + strategy_values = stock_values.sum(axis=1) + cash_amount + n_stocks_series = np.full(len(trading_days), len(tickers)) + + final_total = float(strategy_values[-1]) + if final_total > 0: + new_drifted_allocation = dict(zip(tickers, (stock_values[-1, :] / final_total).tolist())) + else: + new_drifted_allocation = {} + else: + strategy_values = np.full(len(trading_days), portfolio_value) + n_stocks_series = np.zeros(len(trading_days), dtype=int) + final_total = portfolio_value + new_drifted_allocation = {} + + bench_rets = benchmark_returns.reindex(trading_days).fillna(0.0).to_numpy() + benchmark_values = benchmark_value * np.cumprod(1.0 + bench_rets) + + period_df = pd.DataFrame( + {"Strategy": strategy_values, "Benchmark": benchmark_values, "N_Stocks": n_stocks_series}, + index=trading_days, + ) + period_df.index.name = "Date" + + return period_df, final_total, float(benchmark_values[-1]), new_drifted_allocation + + +# ============================================================================= +# 3. ANALYTICS DE PERFORMANCE +# ============================================================================= + +def compute_performance_metrics( + results_df: pd.DataFrame, + rebalance_log: pd.DataFrame, + risk_free_rate: float = RISK_FREE_RATE, +) -> Dict[str, float]: + """ + Calcule le jeu de métriques standard d'un desk quant : + CAGR, volatilité, Sharpe, Sortino, Calmar, Max Drawdown, Alpha/Beta, + Tracking Error, Information Ratio, Win Rate, et coût de turnover moyen. + """ strat = results_df["Strategy"] bench = results_df["Benchmark"] strat_ret = strat.pct_change().dropna() bench_ret = bench.pct_change().dropna() - + n_years = len(strat_ret) / TRADING_DAYS_YEAR cagr = (strat.iloc[-1] / strat.iloc[0]) ** (1 / n_years) - 1 if n_years > 0 else 0.0 vol = strat_ret.std() * np.sqrt(TRADING_DAYS_YEAR) + excess = strat_ret - risk_free_rate / TRADING_DAYS_YEAR sharpe = (excess.mean() / strat_ret.std()) * np.sqrt(TRADING_DAYS_YEAR) if strat_ret.std() != 0 else 0.0 - + downside = strat_ret[strat_ret < 0].std() * np.sqrt(TRADING_DAYS_YEAR) sortino = (cagr - risk_free_rate) / downside if downside > 0 else np.nan - + rolling_max = strat.cummax() drawdown = (strat - rolling_max) / rolling_max max_dd = drawdown.min() - calmar = cagr / abs(max_dd) if max_dd != 0 else np.nan + align_df = pd.concat([strat_ret, bench_ret], axis=1).dropna() - beta = np.cov(align_df.iloc[:, 0], align_df.iloc[:, 1])[0, 1] / np.var(align_df.iloc[:, 1]) if len(align_df) > 1 else np.nan - alpha = (cagr - risk_free_rate) - beta * ((bench.iloc[-1] / bench.iloc[0]) ** (1 / n_years) - 1 - risk_free_rate) - - metrics = { - "CAGR": round(cagr, 4), "Volatility": round(vol, 4), "Sharpe": round(sharpe, 4), - "Sortino": round(sortino, 4), "Calmar": round(calmar, 4), "Max_Drawdown": round(max_dd, 4), - "Alpha": round(alpha, 4), "Beta": round(beta, 4), "Final_Value": round(strat.iloc[-1], 2), + align_df.columns = ["strat", "bench"] + + if len(align_df) > 1 and align_df["bench"].var() > 0: + beta = np.cov(align_df["strat"], align_df["bench"])[0, 1] / np.var(align_df["bench"]) + bench_cagr = (bench.iloc[-1] / bench.iloc[0]) ** (1 / n_years) - 1 if n_years > 0 else 0.0 + alpha = (cagr - risk_free_rate) - beta * (bench_cagr - risk_free_rate) + + active_ret = align_df["strat"] - align_df["bench"] + tracking_error = active_ret.std() * np.sqrt(TRADING_DAYS_YEAR) + information_ratio = ( + (active_ret.mean() * TRADING_DAYS_YEAR) / tracking_error if tracking_error > 0 else np.nan + ) + else: + beta = alpha = tracking_error = information_ratio = np.nan + + win_rate = float((strat_ret > 0).mean()) if len(strat_ret) > 0 else np.nan + avg_turnover = _average_turnover(rebalance_log) + + return { + "CAGR": round(float(cagr), 4), + "Volatility": round(float(vol), 4), + "Sharpe": round(float(sharpe), 4), + "Sortino": round(float(sortino), 4) if pd.notna(sortino) else np.nan, + "Calmar": round(float(calmar), 4) if pd.notna(calmar) else np.nan, + "Max_Drawdown": round(float(max_dd), 4), + "Alpha": round(float(alpha), 4) if pd.notna(alpha) else np.nan, + "Beta": round(float(beta), 4) if pd.notna(beta) else np.nan, + "Tracking_Error": round(float(tracking_error), 4) if pd.notna(tracking_error) else np.nan, + "Information_Ratio": round(float(information_ratio), 4) if pd.notna(information_ratio) else np.nan, + "Win_Rate": round(win_rate, 4) if pd.notna(win_rate) else np.nan, + "Avg_Monthly_Turnover": round(avg_turnover, 4), + "Final_Value": round(float(strat.iloc[-1]), 2), } - return metrics -def backtest_strategy_with_rebalancing(df_daily, df_monthly, model, benchmark_ticker="^FCHI", proba_min=PROBA_MIN, max_stocks=MAX_STOCKS_SELECT): +def _average_turnover(rebalance_log: pd.DataFrame) -> float: + """Turnover moyen d'un mois sur l'autre — proxy du coût de friction réel de la stratégie.""" + if rebalance_log.empty or "Allocation" not in rebalance_log.columns: + return 0.0 + turnovers, prev_alloc = [], {} + for alloc in rebalance_log["Allocation"]: + alloc = alloc or {} + turnovers.append(_compute_turnover(alloc, prev_alloc)) + prev_alloc = alloc + return float(np.mean(turnovers)) if turnovers else 0.0 + + +# ============================================================================= +# 4. API PUBLIQUE +# ============================================================================= + +def backtest_strategy_with_rebalancing( + df_daily: pd.DataFrame, + df_monthly: pd.DataFrame, + model: Any, + benchmark_ticker: str = "^FCHI", + proba_min: float = PROBA_MIN, + max_stocks: int = MAX_STOCKS_SELECT, + conviction_tilt: float = 0.0, +) -> Tuple[pd.DataFrame, pd.DataFrame, Dict[str, float]]: + """ + Backtest complet avec rebalancement mensuel. + + Args: + conviction_tilt: 0.0 = poids Markowitz purs (comportement historique). + >0 incline les poids vers les tickers à plus forte conviction du modèle. + + Returns: + (historique_valorisation, journal_de_rebalancement, métriques_de_performance) + Les métriques sont désormais réellement calculées (CAGR, Sharpe, Alpha, + Information Ratio, turnover moyen, etc.) — elles n'étaient pas renseignées + dans la version précédente du pipeline. + """ df_monthly_feat = add_all_features(df_monthly.copy()) - daily_prices = df_daily["adj close"].unstack().ffill() + daily_prices = _build_price_matrix(df_daily) daily_returns = daily_prices.pct_change().fillna(0) - benchmark_returns = get_benchmark_returns(benchmark_ticker, df_daily.index.get_level_values("date").min(), df_daily.index.get_level_values("date").max(), daily_prices.index) - + benchmark_returns = get_benchmark_returns( + benchmark_ticker, + df_daily.index.get_level_values("date").min(), + df_daily.index.get_level_values("date").max(), + daily_prices.index, + ) + portfolio_value, benchmark_value = 100.0, 100.0 - all_records, rebalance_log, drifted_allocation = [], [], {} + drifted_allocation: Dict[str, float] = {} + period_frames: List[pd.DataFrame] = [] + rebalance_log: List[Dict[str, Any]] = [] + monthly_dates = df_monthly_feat.index.get_level_values("date").unique().sort_values() for i, month_date in enumerate(monthly_dates[:-1]): - month_data = _generate_monthly_signals(df_monthly_feat.xs(month_date, level="date").copy(), model) - allocation = {} + month_data = _generate_monthly_signals( + df_monthly_feat.xs(month_date, level="date").copy(), model + ) + + allocation: Dict[str, float] = {} if not month_data.empty: tickers = _select_tickers(month_data, proba_min, max_stocks) if tickers: - prices_subset = daily_prices[[t for t in tickers if t in daily_prices.columns]].loc[:month_date].iloc[-TRADING_DAYS_YEAR:].dropna(axis=1, thresh=int(TRADING_DAYS_YEAR * 0.8)) + eligible = [t for t in tickers if t in daily_prices.columns] + prices_subset = ( + daily_prices[eligible] + .loc[:month_date] + .iloc[-TRADING_DAYS_YEAR:] + .dropna(axis=1, thresh=int(TRADING_DAYS_YEAR * 0.8)) + ) if not prices_subset.empty: - weights, _ = get_optimal_weights(prices_subset) + weights, method = get_optimal_weights(prices_subset) allocation = {t: w for t, w in weights.items() if w > 1e-4} - - trading_days = daily_prices.index[(daily_prices.index >= month_date) & (daily_prices.index < monthly_dates[i+1])] - day_records, portfolio_value, benchmark_value, drifted_allocation = _simulate_daily_returns(allocation, drifted_allocation, trading_days, daily_returns, benchmark_returns, portfolio_value, benchmark_value) - all_records.extend(day_records) + if conviction_tilt > 0: + allocation = _blend_with_conviction( + allocation, month_data["proba_upside"], conviction_tilt + ) + logger.debug(f"[{month_date.date()}] Optimisation via '{method}', {len(allocation)} positions.") + + trading_days = daily_prices.index[ + (daily_prices.index >= month_date) & (daily_prices.index < monthly_dates[i + 1]) + ] + period_df, portfolio_value, benchmark_value, drifted_allocation = _simulate_period( + allocation, drifted_allocation, trading_days, daily_returns, benchmark_returns, + portfolio_value, benchmark_value, + ) + period_frames.append(period_df) rebalance_log.append({"Date": month_date, "N_Stocks": len(allocation), "Allocation": allocation}) - return pd.DataFrame(all_records).set_index("Date"), pd.DataFrame(rebalance_log).set_index("Date"), {} + hist_df = pd.concat(period_frames) if period_frames else pd.DataFrame(columns=["Strategy", "Benchmark", "N_Stocks"]) + rebal_df = pd.DataFrame(rebalance_log).set_index("Date") if rebalance_log else pd.DataFrame() + metrics = compute_performance_metrics(hist_df, rebal_df) if not hist_df.empty else {} -def generate_live_signals(df_daily, daily_prices, model, rebalance_history, proba_min=PROBA_MIN, max_stocks=MAX_STOCKS_SELECT): + return hist_df, rebal_df, metrics + + +def generate_live_signals( + df_daily: pd.DataFrame, + daily_prices: pd.DataFrame, + model: Any, + rebalance_history: pd.DataFrame, + proba_min: float = PROBA_MIN, + max_stocks: int = MAX_STOCKS_SELECT, + conviction_tilt: float = 0.0, +) -> Tuple[pd.DataFrame, pd.DataFrame]: + """ + Génère les signaux du jour (séance N) pour alimenter le dashboard. + + Fix critique : `_build_daily_snapshot` renvoie désormais un index simple + de tickers (via `.xs(level="date")`) au lieu d'un MultiIndex (ticker, date). + Avant ce fix, `_select_tickers` retournait des tuples `(ticker, date)` qui ne + matchaient jamais `daily_prices.columns`, laissant l'allocation vide en + permanence — d'où les signaux NEUTRAL à 0.00 malgré des probabilités élevées. + """ snapshot, last_date = _build_daily_snapshot(df_daily) snapshot_scored = _generate_monthly_signals(snapshot, model) - - # 1. Allocation dynamique quotidienne + + if snapshot_scored.empty: + logger.error("Snapshot scoré vide — aucun signal ne peut être généré aujourd'hui.") + empty_signals = pd.DataFrame(columns=["Ticker", "Signal", "Allocation", "Proba_Hausse"]) + return empty_signals, rebalance_history + tickers = _select_tickers(snapshot_scored, proba_min, max_stocks) - allocation = {} + allocation: Dict[str, float] = {} if tickers: - prices_subset = daily_prices[[t for t in tickers if t in daily_prices.columns]].loc[:last_date].iloc[-TRADING_DAYS_YEAR:].dropna(axis=1, thresh=int(TRADING_DAYS_YEAR * 0.8)) + eligible = [t for t in tickers if t in daily_prices.columns] + if len(eligible) < len(tickers): + missing = set(tickers) - set(eligible) + logger.warning(f"{len(missing)} ticker(s) sélectionné(s) absents de la matrice de prix : {missing}") + + prices_subset = ( + daily_prices[eligible] + .loc[:last_date] + .iloc[-TRADING_DAYS_YEAR:] + .dropna(axis=1, thresh=int(TRADING_DAYS_YEAR * 0.8)) + ) if not prices_subset.empty: - weights, _ = get_optimal_weights(prices_subset) + weights, method = get_optimal_weights(prices_subset) allocation = {t: w for t, w in weights.items() if w > 1e-4} - - # 2. Log mensuel pour ton dashboard + if conviction_tilt > 0: + allocation = _blend_with_conviction( + allocation, snapshot_scored["proba_upside"], conviction_tilt + ) + logger.info(f"Allocation du jour ({method}) : {len(allocation)} positions sur {len(eligible)} candidats.") + else: + logger.warning("Historique de prix insuffisant pour les tickers sélectionnés — allocation vide.") + else: + logger.info(f"Aucun ticker au-dessus du seuil de probabilité ({proba_min:.0%}) aujourd'hui.") + + # Log mensuel pour le dashboard (un seul rebalancement enregistré par mois) last_rebalance_date = rebalance_history.index.max() if not rebalance_history.empty else None if last_rebalance_date is None or (last_date.year, last_date.month) != (last_rebalance_date.year, last_rebalance_date.month): new_row = pd.DataFrame([{"N_Stocks": len(allocation), "Allocation": allocation}], index=[last_date]) @@ -228,17 +554,25 @@ def generate_live_signals(df_daily, daily_prices, model, rebalance_history, prob rebalance_history = pd.concat([rebalance_history, new_row]).sort_index() out = snapshot_scored.reset_index().rename(columns={"ticker": "Ticker"}) - - # NORMALISATION : on retire le suffixe du ticker du DF pour correspondre aux clés du dictionnaire[cite: 1] - out["Ticker_Short"] = out["Ticker"].apply(lambda x: x.split('.')[0]) - out["Allocation"] = out["Ticker_Short"].map(allocation).fillna(0.0) - + out["ticker_root"] = out["Ticker"].apply(lambda t: t.split(".", 1)[0]) + out["Allocation"] = out["ticker_root"].map(allocation).fillna(0.0) out["Signal"] = np.where(out["Allocation"] > 0, "BUY", "NEUTRAL") out["Proba_Hausse"] = (out["proba_upside"] * 100).round(1) - + return out[["Ticker", "Signal", "Allocation", "Proba_Hausse"]], rebalance_history -def _build_daily_snapshot(df_daily): + +def _build_daily_snapshot(df_daily: pd.DataFrame) -> Tuple[pd.DataFrame, pd.Timestamp]: + """ + Construit le snapshot de la dernière séance, avec un index simple de tickers. + + Avant : filtrage booléen sur un MultiIndex (ticker, date), qui conserve + le MultiIndex complet dans le résultat. + Après : `.xs(level="date")`, identique à la logique déjà utilisée (et + fonctionnelle) dans le backtest — aligne le comportement live sur le + comportement backtesté. + """ last_date = df_daily.index.get_level_values("date").max() df_feat = add_all_features(df_daily.copy()) - return df_feat[df_feat.index.get_level_values("date") == last_date].copy(), last_date + snapshot = df_feat.xs(last_date, level="date").copy() + return snapshot, last_date \ No newline at end of file From 950f81f1cdd058dd7339ccc9eb5ced7add9e3680 Mon Sep 17 00:00:00 2001 From: MoussaTheAnalyst <163172464+SORADATA@users.noreply.github.com> Date: Tue, 7 Jul 2026 11:36:17 +0200 Subject: [PATCH 03/17] Fix: daily snapshot index and allocation mapping Fix critical issues with daily snapshot and allocation mapping to ensure correct signal generation. --- src/pipeline/backtest.py | 38 +++++++++++++++++++++++++++++++++----- 1 file changed, 33 insertions(+), 5 deletions(-) diff --git a/src/pipeline/backtest.py b/src/pipeline/backtest.py index 72a3035..3553ca1 100644 --- a/src/pipeline/backtest.py +++ b/src/pipeline/backtest.py @@ -505,11 +505,21 @@ def generate_live_signals( """ Génère les signaux du jour (séance N) pour alimenter le dashboard. - Fix critique : `_build_daily_snapshot` renvoie désormais un index simple + Fix critique #1 (hérité) : `_build_daily_snapshot` renvoie un index simple de tickers (via `.xs(level="date")`) au lieu d'un MultiIndex (ticker, date). Avant ce fix, `_select_tickers` retournait des tuples `(ticker, date)` qui ne matchaient jamais `daily_prices.columns`, laissant l'allocation vide en - permanence — d'où les signaux NEUTRAL à 0.00 malgré des probabilités élevées. + permanence — d'où des signaux NEUTRAL malgré des probabilités élevées. + + Fix critique #2 (ce correctif) : le mapping final Ticker -> Allocation + utilisait une clé "ticker_root" (suffixe de place type ".PA"/".AS" retiré, + ex "AI.PA" -> "AI"), alors que le dict `allocation` est construit avec les + tickers COMPLETS (même format que `daily_prices.columns`, ex "AI.PA"). + Le `.map(allocation)` sur la version tronquée ne trouvait donc jamais de + correspondance -> Allocation = 0.0 pour toutes les lignes -> Signal = + "NEUTRAL" systématique, même quand `proba_upside` dépassait largement le + seuil `proba_min`. Le mapping se fait désormais directement sur `Ticker`, + dans le même référentiel que `allocation`. """ snapshot, last_date = _build_daily_snapshot(df_daily) snapshot_scored = _generate_monthly_signals(snapshot, model) @@ -554,8 +564,26 @@ def generate_live_signals( rebalance_history = pd.concat([rebalance_history, new_row]).sort_index() out = snapshot_scored.reset_index().rename(columns={"ticker": "Ticker"}) - out["ticker_root"] = out["Ticker"].apply(lambda t: t.split(".", 1)[0]) - out["Allocation"] = out["ticker_root"].map(allocation).fillna(0.0) + + # --- Mapping Allocation --------------------------------------------- + # `allocation` est keyé avec le ticker COMPLET (même format que + # `daily_prices.columns`, ex "AI.PA"). On mappe donc directement sur + # `Ticker`, sans tronquer le suffixe de place — ce tronquage était la + # cause du bug (voir docstring ci-dessus). + out["Allocation"] = out["Ticker"].map(allocation).fillna(0.0) + + # Garde-fou : si `allocation` n'est pas vide mais qu'aucune ligne n'a pu + # être mappée, c'est très probablement un nouveau mismatch de format de + # ticker entre `snapshot_scored.index` et `daily_prices.columns`. + if allocation and out["Allocation"].sum() == 0.0: + logger.warning( + "Allocation non nulle calculée (%d positions) mais aucun mapping " + "vers 'Ticker' n'a matché — vérifier la cohérence de format entre " + "snapshot_scored.index et daily_prices.columns (ex : suffixe de " + "place manquant/en trop).", + len(allocation), + ) + out["Signal"] = np.where(out["Allocation"] > 0, "BUY", "NEUTRAL") out["Proba_Hausse"] = (out["proba_upside"] * 100).round(1) @@ -575,4 +603,4 @@ def _build_daily_snapshot(df_daily: pd.DataFrame) -> Tuple[pd.DataFrame, pd.Time last_date = df_daily.index.get_level_values("date").max() df_feat = add_all_features(df_daily.copy()) snapshot = df_feat.xs(last_date, level="date").copy() - return snapshot, last_date \ No newline at end of file + return snapshot, last_date From b8d292e922486f6500a171279005ce5e03bb4ded Mon Sep 17 00:00:00 2001 From: MoussaTheAnalyst <163172464+SORADATA@users.noreply.github.com> Date: Tue, 7 Jul 2026 12:13:11 +0200 Subject: [PATCH 04/17] Refactor: MLflow logging and improve error handling --- src/models/train.py | 269 ++++++-------------------------------------- 1 file changed, 36 insertions(+), 233 deletions(-) diff --git a/src/models/train.py b/src/models/train.py index 082c809..479512f 100644 --- a/src/models/train.py +++ b/src/models/train.py @@ -32,102 +32,30 @@ mlflow.set_experiment("AlphaEdge_Ensemble_Production") logger.info(f"MLflow activé — tracking URI : {mlflow.get_tracking_uri()}") else: - logger.warning("HF_TOKEN absent — MLflow désactivé, entraînement local uniquement.") + logger.warning("HF_TOKEN absent — MLflow désactivé.") - -# ============================================================================= -# HELPERS MLFLOW -# ============================================================================= def _check_registry_available(client: MlflowClient) -> bool: - """ - Vérifie que le backend MLflow distant supporte le Model Registry - avant de tenter un register_model (sinon erreur explicite et claire). - """ try: client.search_registered_models(max_results=1) return True - except MlflowException as e: - logger.error( - "Le Model Registry ne semble pas disponible sur le serveur MLflow distant " - f"(backend-store probablement non compatible) : {e}" - ) + except MlflowException: return False - -# ============================================================================= -# WALK-FORWARD VALIDATION -# ============================================================================= -def walk_forward_eval( - df: pd.DataFrame, - n_windows: int = 4, - test_months: int = 3, - n_optuna_trials: int = 20, -) -> pd.DataFrame: - dates = df.index.get_level_values("date").unique().sort_values() - results = [] - - for i in range(n_windows): - test_end = dates[-(i * test_months + 1)] - test_start = dates[-(i * test_months + test_months)] - train_end = test_start - pd.DateOffset(months=1) - - df_tr = df[df.index.get_level_values("date") <= train_end] - df_te = df[ - (df.index.get_level_values("date") >= test_start) - & (df.index.get_level_values("date") <= test_end) - ] - - if len(df_tr) < 50 or len(df_te) < 5: - continue - if len(df_te["target"].unique()) < 2: - continue - - model = AlphaEdgeEnsemble(n_optuna_trials=n_optuna_trials) - model.fit(df_tr, df_tr["target"]) - proba = model.predict_proba(df_te)[:, 1] - - results.append({ - "window": i + 1, - "test_start": str(test_start.date()), - "test_end": str(test_end.date()), - "auc": round(roc_auc_score(df_te["target"], proba), 4), - "apr": round(average_precision_score(df_te["target"], proba), 4), - "n_test": len(df_te), - }) - logger.info( - f"Window {i+1} | AUC: {results[-1]['auc']:.4f} | " - f"APR: {results[-1]['apr']:.4f} | n={results[-1]['n_test']}" - ) - - return pd.DataFrame(results) - - -# ============================================================================= -# WRAPPER MLFLOW PYFUNC -# ============================================================================= class AlphaEdgePyFunc(mlflow.pyfunc.PythonModel): def __init__(self, model_instance): self.model = model_instance def predict(self, context, model_input, params=None): - # MLflow s'attend à recevoir un DataFrame en entrée - # On retourne uniquement la probabilité de la classe 1 (le signal d'achat) return self.model.predict_proba(model_input)[:, 1] - -# ============================================================================= -# PIPELINE D'ENTRAÎNEMENT -# ============================================================================= def train_pipeline(market_name: str) -> tuple[AlphaEdgeEnsemble, dict]: - logger.info(f"Début du cycle d'entraînement AlphaEdge — {market_name}") + logger.info(f"Début entraînement — {market_name}") data_path = DATA_DIR / "processed" / market_name / "monthly_features.parquet" if not data_path.exists(): - raise FileNotFoundError(f"Fichier source introuvable : {data_path}") + raise FileNotFoundError(f"Source introuvable : {data_path}") df = pd.read_parquet(data_path) - - # Eviter le leakage data df["future_return"] = df.groupby(level="ticker")["adj close"].pct_change(1).shift(-1) df["target"] = df["future_return"].gt(0).astype(int) df = df.dropna(subset=["target", "future_return"]) @@ -135,201 +63,76 @@ def train_pipeline(market_name: str) -> tuple[AlphaEdgeEnsemble, dict]: dates = df.index.get_level_values("date") split_date = dates.max() - pd.DateOffset(months=6) - # Split temporel AVANT add_all_features pour éviter le leakage global - df_train = df[dates <= split_date].copy() - df_test = df[dates > split_date].copy() - - df_train = add_all_features(df_train) - df_test = add_all_features(df_test) + df_train = add_all_features(df[dates <= split_date].copy()) + df_test = add_all_features(df[dates > split_date].copy()) - if len(df_train) < 100: - raise ValueError(f"Volume de données insuffisant : {len(df_train)}") - - # Baseline - all_feats = [f for g in FEATURE_GROUPS.values() for f in g] - available = [f for f in all_feats if f in df_train.columns] - dummy = DummyClassifier(strategy="most_frequent").fit(df_train[available], df_train["target"]) - baseline_auc = roc_auc_score(df_test["target"], dummy.predict_proba(df_test[available])[:, 1]) - - # Entraînement model = AlphaEdgeEnsemble(n_optuna_trials=50) model.fit(df_train, df_train["target"]) proba = model.predict_proba(df_test)[:, 1] final_auc = roc_auc_score(df_test["target"], proba) final_apr = average_precision_score(df_test["target"], proba) - lift = final_auc - baseline_auc fin_metrics = calculate_financial_metrics(df_test, probas=proba, threshold=0.5) - logger.info(f"ML -> AUC : {final_auc:.4f} | APR : {final_apr:.4f} | Lift : +{lift:.4f}") - logger.info(f"Finances -> Sharpe : {fin_metrics['sharpe']} | Max DD : {fin_metrics['max_drawdown']}") - - # Walk-forward sur df entier (target déjà calculé, pas de leakage supplémentaire) - df_full = pd.concat([df_train, df_test]) - wf_results = walk_forward_eval(df_full, n_windows=4, test_months=3, n_optuna_trials=20) - - # Sauvegarde locale par marché (TOUJOURS faite, indépendamment de MLflow) market_model_dir = MODEL_DIR / market_name market_model_dir.mkdir(parents=True, exist_ok=True) model.save(market_model_dir / "ensemble_model.pkl") - logger.info(f"Modèle sauvegardé : {market_model_dir / 'ensemble_model.pkl'}") model_card = { - "market": market_name, - "trained_at": pd.Timestamp.now().isoformat(), - "architecture": "XGBoost + LightGBM + Ridge → LogisticRegression", - "metrics_ml": { - "auc_test": round(final_auc, 4), - "apr_test": round(final_apr, 4), - "baseline_auc": round(baseline_auc, 4), - "lift": round(lift, 4), - }, - "metrics_fin": fin_metrics, - "model_weights": model.get_model_weights(), - "walk_forward": wf_results.to_dict(orient="records") if not wf_results.empty else [], - "n_features": len(model.features_), - "train_size": len(df_train), - "test_size": len(df_test), - "mlflow": { - "attempted": USE_MLFLOW, - "success": False, - "run_id": None, - "version": None, - "promoted": False, - }, + "market": market_name, + "metrics_ml": {"auc_test": round(final_auc, 4), "apr_test": round(final_apr, 4)}, + "metrics_fin": fin_metrics, + "mlflow": {"success": False, "run_id": None, "promoted": False}, } - with open(market_model_dir / "model_card.json", "w") as f: - json.dump(model_card, f, indent=2) - # ========================================================================= - # MLFLOW — chaque étape est tracée, aucune exception n'est avalée silencieusement - # ========================================================================= if USE_MLFLOW: registered_model_name = f"AlphaEdge_Ensemble_{market_name}" client = MlflowClient() - if not _check_registry_available(client): - logger.error( - f"[{market_name}] Model Registry indisponible — le run sera loggé " - "mais ne sera PAS enregistré dans le registre." - ) - try: with mlflow.start_run(run_name=f"Ensemble_{market_name}") as run: - mlflow.log_params({ - "architecture": "XGB+LGB+Ridge->LR", - "market": market_name, - "n_features": len(model.features_), - }) mlflow.log_metrics({ - "AUC_Test": final_auc, - "APR_Test": final_apr, + "AUC_Test": final_auc, "Sharpe_Ratio": fin_metrics["sharpe"], - "Max_Drawdown": fin_metrics["max_drawdown"], - "Total_Return": fin_metrics["total_return"], - "WF_AUC_mean": wf_results["auc"].mean() if not wf_results.empty else 0.0, - "primary_metric": final_auc + "Max_Drawdown": fin_metrics["max_drawdown"] }) - mlflow.log_dict(model_card, "model_card.json") - - # --- Création de l'instance PyFunc et de la Signature --- pyfunc_model = AlphaEdgePyFunc(model_instance=model) - - input_example = df_test[model.features_].head(5) - signature = infer_signature( - model_input=input_example, - model_output=pyfunc_model.predict(context=None, model_input=input_example) - ) - - model_info = mlflow.pyfunc.log_model( - artifact_path="ensemble_model", - python_model=pyfunc_model, - signature=signature, - input_example=input_example - ) - - logger.info(f"[{market_name}] Modèle loggé sur MLflow — run_id={run.info.run_id}") + mlflow.pyfunc.log_model("ensemble_model", python_model=pyfunc_model) - model_card["mlflow"]["success"] = True - model_card["mlflow"]["run_id"] = run.info.run_id + model_card["mlflow"].update({"success": True, "run_id": run.info.run_id}) - # Enregistrement dans le Model Registry + mv = mlflow.register_model(f"runs:/{run.info.run_id}/ensemble_model", registered_model_name) + + champion_sharpe = -999.0 try: - mv = mlflow.register_model( - model_uri=f"runs:/{run.info.run_id}/ensemble_model", - name=registered_model_name, - ) - model_card["mlflow"]["version"] = mv.version - logger.info( - f"[{market_name}] Modèle enregistré : " - f"{registered_model_name} v{mv.version}" - ) - - if ( - fin_metrics["sharpe"] >= SHARPE_THRESHOLD - and fin_metrics["max_drawdown"] >= MAX_DD_THRESHOLD - ): - client.set_registered_model_alias( - registered_model_name, "champion", mv.version - ) + current_champ = client.get_model_version_by_alias(registered_model_name, "champion") + champion_sharpe = client.get_run(current_champ.run_id).data.metrics.get("Sharpe_Ratio", -999.0) + except: + logger.info(f"[{market_name}] Premier modèle détecté.") + + if fin_metrics["sharpe"] >= SHARPE_THRESHOLD and fin_metrics["max_drawdown"] >= MAX_DD_THRESHOLD: + if fin_metrics["sharpe"] >= champion_sharpe: + client.set_registered_model_alias(registered_model_name, "champion", mv.version) model_card["mlflow"]["promoted"] = True - logger.info( - f"[{market_name}] Promotion : Version {mv.version} → 'champion'." - ) + logger.info(f"[{market_name}] Promotion réussie : v{mv.version}") else: - logger.warning( - f"[{market_name}] Promotion refusée : seuils non atteints " - f"(Sharpe={fin_metrics['sharpe']:.2f}, " - f"MaxDD={fin_metrics['max_drawdown']:.2f}). " - "Ancien champion maintenu." - ) - except MlflowException: - logger.exception( - f"[{market_name}] Échec de l'enregistrement dans le Model Registry " - "— le run reste loggé mais le modèle n'est pas versionné." - ) + logger.warning(f"[{market_name}] Challenger battu par Champion.") + else: + logger.warning(f"[{market_name}] Seuils sécurité non atteints.") - except MlflowException: - logger.exception( - f"[{market_name}] Échec MLflow (erreur API/connexion) — " - "modèle local sauvegardé uniquement." - ) - except Exception: - logger.exception( - f"[{market_name}] Échec MLflow (erreur inattendue) — " - "modèle local sauvegardé uniquement." - ) + except Exception as e: + logger.error(f"Erreur MLflow : {e}") - # Re-sauvegarde du model_card avec le statut MLflow final - with open(market_model_dir / "model_card.json", "w") as f: - json.dump(model_card, f, indent=2) + with open(market_model_dir / "model_card.json", "w") as f: + json.dump(model_card, f, indent=2) return model, model_card - -# ============================================================================= -# ORCHESTRATEUR -# ============================================================================= if __name__ == "__main__": config_dir = CONFIG_DIR / "markets" - if not config_dir.exists(): - raise FileNotFoundError(f"Dossier de configs introuvable : {config_dir}") - - failures = [] for config_file in sorted(config_dir.glob("*.json")): with open(config_file) as f: - market_cfg = json.load(f) - market = market_cfg.get("market_name") - if not market: - logger.warning(f"Clé 'market_name' absente dans {config_file.name}, ignoré.") - continue - try: - train_pipeline(market) - except Exception: - logger.exception(f"[{market}] Échec complet du pipeline d'entraînement.") - failures.append(market) - - if failures: - logger.error(f"Marchés en échec : {failures}") - raise SystemExit(1) + market = json.load(f).get("market_name") + if market: + train_pipeline(market) From 4a288f5a127320489e5235f7a0ba9627683cf2bf Mon Sep 17 00:00:00 2001 From: MoussaTheAnalyst <163172464+SORADATA@users.noreply.github.com> Date: Tue, 7 Jul 2026 14:02:07 +0200 Subject: [PATCH 05/17] feat: Add benchmark_ticker to CAC40 market configuration --- config/markets/cac40.json | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/config/markets/cac40.json b/config/markets/cac40.json index fb1f9da..21f6879 100644 --- a/config/markets/cac40.json +++ b/config/markets/cac40.json @@ -1,5 +1,6 @@ { "market_name": "CAC40", + "benchmark_ticker": "^FCHI", "tickers": [ "AI.PA", "AIR.PA", "ALO.PA", "MT.AS", "ATO.PA", "CS.PA", "BNP.PA", "EN.PA", "CAP.PA", "CA.PA", "DSY.PA", "EL.PA", "ENGI.PA", "ERF.PA", @@ -9,4 +10,4 @@ "URW.PA", "VIE.PA", "DG.PA", "VIV.PA", "WLN.PA", "FR.PA" ], "ff_region": "Europe_5_Factors" -} \ No newline at end of file +} From f332ee519afd9d17b0ee0e2228a08f4ca8d521a2 Mon Sep 17 00:00:00 2001 From: MoussaTheAnalyst <163172464+SORADATA@users.noreply.github.com> Date: Tue, 7 Jul 2026 14:28:04 +0200 Subject: [PATCH 06/17] Refactor: daily_run.py to improve structure and clarity --- src/pipeline/daily_run.py | 81 +++++++++++++++++++-------------------- 1 file changed, 39 insertions(+), 42 deletions(-) diff --git a/src/pipeline/daily_run.py b/src/pipeline/daily_run.py index 19f0f72..5ef0c2b 100644 --- a/src/pipeline/daily_run.py +++ b/src/pipeline/daily_run.py @@ -9,11 +9,14 @@ from src.utils.logger import setup_logger from src.pipeline.etl import get_data_pipeline -from src.pipeline.backtest import backtest_strategy_with_rebalancing, generate_live_signals +from src.pipeline.backtest import ( + backtest_strategy_with_rebalancing, + generate_live_signals, + _build_price_matrix +) from src.models.model_loader import load_champion from const import BACKTEST_YEARS - # ============================================================================= # CONFIGURATION GLOBALE # ============================================================================= @@ -22,7 +25,6 @@ HF_REPO_ID = os.getenv("HF_REPO_ID", "soradata/alphaedge-data") hf_api = HfApi() - def upload_to_hf(local_path: Path, hf_filename: str, market_name: str) -> bool: """Upload vers le repo Hugging Face, sous data/{market_name}/{hf_filename}.""" if not HF_TOKEN: @@ -40,14 +42,13 @@ def upload_to_hf(local_path: Path, hf_filename: str, market_name: str) -> bool: print(f"Erreur Upload HF ({market_name}/{hf_filename}): {e}") return False - def load_rebalance_history_from_hf(market_name: str, local_fallback: Path) -> pd.DataFrame: """ Récupère l'historique de rebalancing existant (source de vérité pour savoir si un nouveau rebalancing mensuel doit être déclenché). Fallback sur le fichier local, puis sur un DataFrame vide. """ - empty = pd.DataFrame(columns=["N_Stocks", "Optim_Method", "Allocation", "Top_Ticker"]) + empty = pd.DataFrame(columns=["N_Stocks", "Allocation"]) empty.index.name = "Date" if HF_TOKEN: @@ -61,8 +62,8 @@ def load_rebalance_history_from_hf(market_name: str, local_fallback: Path) -> pd df = pd.read_parquet(path) df.index = pd.to_datetime(df.index) return df - except Exception as e: - print(f"Impossible de charger rebalance_history depuis HF ({market_name}): {e}") + except Exception: + pass if local_fallback.exists(): try: @@ -74,7 +75,6 @@ def load_rebalance_history_from_hf(market_name: str, local_fallback: Path) -> pd return empty - # ============================================================================= # PIPELINE PRINCIPAL (daily run) # ============================================================================= @@ -84,7 +84,7 @@ def run_pipeline(market_config: dict) -> None: logger.info(f"STARTING DAILY PIPELINE | MARKET: {market_name}") try: - # 1. Chargement du modèle champion (MLflow pyfunc -> fallback local .pkl) + # 1. Chargement du modèle champion model = load_champion(market_name) logger.info(f"Modèle champion prêt pour {market_name} | type={type(model).__name__}") @@ -101,10 +101,7 @@ def run_pipeline(market_config: dict) -> None: raise ValueError(f"Pas de données sur les {BACKTEST_YEARS} dernières années.") last_session = df_daily_bt.index.get_level_values("date").max() - logger.info( - f"Dernière séance récupérée : {last_session.date()} | " - f"Backtest window : {cutoff.date()} → aujourd'hui" - ) + logger.info(f"Dernière séance récupérée : {last_session.date()} | Backtest window : {cutoff.date()} -> aujourd'hui") base_dir = Path(f"data/processed/{market_name}") base_dir.mkdir(parents=True, exist_ok=True) @@ -113,30 +110,35 @@ def run_pipeline(market_config: dict) -> None: signals_path = base_dir / "latest_signals.parquet" metadata_path = base_dir / "data_metadata.json" - # 3. Backtest historique (référence de performance, alimente aussi le dashboard) - logger.info(f"Executing backtest for {market_name}...") + # Création standardisée de la matrice de prix + daily_prices = _build_price_matrix(df_daily_bt) + bench_ticker = market_config.get("benchmark_ticker") + + if not bench_ticker: + raise ValueError(f"Aucun benchmark_ticker configuré pour {market_name}.") + + # 3. Backtest historique + logger.info(f"Executing backtest for {market_name} with benchmark {bench_ticker}...") hist_df, rebal_df_backtest, metrics = backtest_strategy_with_rebalancing( df_daily_bt, df_monthly_bt, model, - benchmark_ticker=market_config.get("benchmark_ticker", "^FCHI"), + benchmark_ticker=bench_ticker, ) - # 4. Signaux live (séance N) — proba quotidienne, allocation gelée - # jusqu'au prochain rebalancing mensuel réel. + # 4. Signaux live (séance N) logger.info(f"Generating daily live signals for {market_name} (session {last_session.date()})...") rebalance_history = load_rebalance_history_from_hf(market_name, rebal_path) - daily_prices = df_daily_bt["adj close"].unstack().ffill() - + signals_df, rebalance_history_updated = generate_live_signals( - df_daily_bt, daily_prices, model, rebalance_history, + df_daily_bt, + daily_prices, + model, + rebalance_history, ) if signals_df.empty: - logger.error( - f"[{market_name}] ATTENTION : latest_signals est vide ! " - "Vérifier les logs de scoring ci-dessus (compatibilité modèle pyfunc/natif)." - ) + logger.error(f"[{market_name}] ATTENTION : latest_signals est vide. Vérifier les logs de scoring.") # 5. Sauvegarde locale hist_df.to_parquet(hist_path) @@ -144,16 +146,16 @@ def run_pipeline(market_config: dict) -> None: signals_df.to_parquet(signals_path) metadata = { - "market_name": market_name, - "last_run_utc": datetime.now(timezone.utc).isoformat(), + "market_name": market_name, + "last_run_utc": datetime.now(timezone.utc).isoformat(), "last_session_date": str(last_session.date()), - "model_type": type(model).__name__, - "n_signals": len(signals_df), - "n_buy_signals": int((signals_df["Signal"] == "BUY").sum()) if not signals_df.empty else 0, - "last_rebalance": str(rebalance_history_updated.index.max().date()) - if not rebalance_history_updated.empty else None, - "metrics": metrics, + "model_type": type(model).__name__, + "n_signals": len(signals_df), + "n_buy_signals": int((signals_df["Signal"] == "BUY").sum()) if not signals_df.empty else 0, + "last_rebalance": str(rebalance_history_updated.index.max().date()) if not rebalance_history_updated.empty else None, + "metrics": metrics, } + with open(metadata_path, "w") as f: json.dump(metadata, f, indent=2, default=str) @@ -163,8 +165,8 @@ def run_pipeline(market_config: dict) -> None: upload_results = { "portfolio_history": upload_to_hf(hist_path, "portfolio_history.parquet", market_name), "rebalance_history": upload_to_hf(rebal_path, "rebalance_history.parquet", market_name), - "latest_signals": upload_to_hf(signals_path, "latest_signals.parquet", market_name), - "data_metadata": upload_to_hf(metadata_path, "data_metadata.json", market_name), + "latest_signals": upload_to_hf(signals_path, "latest_signals.parquet", market_name), + "data_metadata": upload_to_hf(metadata_path, "data_metadata.json", market_name), } failed_uploads = [k for k, ok in upload_results.items() if not ok] @@ -175,17 +177,12 @@ def run_pipeline(market_config: dict) -> None: else: logger.info("Synchronisation Hugging Face terminée avec succès.") - logger.info( - f"Pipeline terminé | Sharpe: {metrics.get('Sharpe', 'N/A')} | " - f"Signaux BUY: {metadata['n_buy_signals']}/{metadata['n_signals']} | " - f"Dernier rebalancing: {metadata['last_rebalance']}" - ) + logger.info(f"Pipeline terminé | Sharpe: {metrics.get('Sharpe', 'N/A')} | Signaux BUY: {metadata['n_buy_signals']}/{metadata['n_signals']}") except Exception as e: logger.critical(f"CRITICAL FAILURE {market_name}: {e}", exc_info=True) raise - # ============================================================================= # ORCHESTRATEUR # ============================================================================= @@ -204,4 +201,4 @@ def run_pipeline(market_config: dict) -> None: if failures: print(f"Marchés en échec : {failures}") - raise SystemExit(1) \ No newline at end of file + raise SystemExit(1) From 414f787b10f5237c21b1cc7912b27134574ad581 Mon Sep 17 00:00:00 2001 From: MoussaTheAnalyst <163172464+SORADATA@users.noreply.github.com> Date: Tue, 7 Jul 2026 14:35:50 +0200 Subject: [PATCH 07/17] Refactor: backtest for clarity and functionality Refactor backtest.py to improve code structure and clarity. Update function signatures, add new features, and enhance logging. --- src/pipeline/backtest.py | 188 ++++++++------------------------------- 1 file changed, 36 insertions(+), 152 deletions(-) diff --git a/src/pipeline/backtest.py b/src/pipeline/backtest.py index 3553ca1..002f615 100644 --- a/src/pipeline/backtest.py +++ b/src/pipeline/backtest.py @@ -9,18 +9,8 @@ 1. Construction de portefeuille : sélection des tickers + optimisation Markowitz 2. Moteur de simulation : vectorisé (numpy), pas de boucle Python par jour 3. Analytics de performance : Sharpe, Sortino, Calmar, Alpha/Beta, Information Ratio, - Tracking Error, coût de turnover + Tracking Error, coût de turnover 4. API publique : backtest_strategy_with_rebalancing / generate_live_signals - -Notes de conception --------------------- -- Les positions non allouées (somme des poids < 1) sont conservées en cash à - rendement nul, explicitement comptabilisées — pas de "cash fantôme". -- Les coûts de transaction sont appliqués à chaque rebalancement, proportionnellement - au turnover réel entre l'allocation cible et l'allocation *drifted* (dérivée - des variations de prix depuis le dernier rebalancement). -- Aucune métrique de performance passée ne constitue une garantie de résultats - futurs. Ce module sert à l'analyse quantitative, pas à du conseil en investissement. """ from __future__ import annotations @@ -47,15 +37,9 @@ logger = setup_logger("backtest") -# Nombre de jours de forward-fill toléré pour un prix manquant avant de -# considérer le titre comme illiquide/délisté et de l'exclure de la fenêtre. MAX_PRICE_FFILL_DAYS = 5 - -# Seuil de features manquantes (moyenne sur toutes les colonnes) au-delà -# duquel on log un warning explicite plutôt que d'imputer silencieusement. MAX_ACCEPTABLE_MISSING_RATIO = 0.05 - # ============================================================================= # 1. CONSTRUCTION DE PORTEFEUILLE # ============================================================================= @@ -64,17 +48,6 @@ def get_optimal_weights( prices_df: pd.DataFrame, risk_free_rate: float = RISK_FREE_RATE, ) -> Tuple[Dict[str, float], str]: - """ - Optimise les poids du portefeuille (Max Sharpe -> Min Vol -> Equal Weight en cascade). - - Args: - prices_df: Prix ajustés, colonnes = tickers, index = dates. - risk_free_rate: Taux sans risque annualisé utilisé pour Max Sharpe. - - Returns: - (weights, method) où method indique la stratégie effectivement utilisée, - utile pour le monitoring de la robustesse de l'optimiseur en production. - """ n_assets = prices_df.shape[1] if n_assets < MIN_STOCKS_OPTIM: return {t: 1.0 / n_assets for t in prices_df.columns}, "equal_weight" @@ -110,7 +83,6 @@ def get_optimal_weights( def _log_concentration(weights: Dict[str, float]) -> None: - """Log l'indice de Herfindahl-Hirschman du portefeuille (mesure de concentration).""" active = np.array([w for w in weights.values() if w > 1e-6]) if active.size == 0: return @@ -119,14 +91,6 @@ def _log_concentration(weights: Dict[str, float]) -> None: def _score_with_model(model: Any, features: pd.DataFrame) -> np.ndarray: - """ - Scoring robuste du modèle avec reindex pour gérer les colonnes manquantes - (ex : lags macro non disponibles le jour J). - - Toute imputation par zéro est loggée si elle dépasse un seuil raisonnable, - afin de détecter une dérive de la qualité de données en production plutôt - que de la masquer silencieusement. - """ if hasattr(model, "predict_proba"): expected_cols = getattr(model, "features_", None) x_input = features.reindex(columns=expected_cols).copy() if expected_cols else features.copy() @@ -158,10 +122,29 @@ def _warn_if_missing(x_input: pd.DataFrame) -> None: f"Scoring : {missing_ratio:.1%} de valeurs manquantes en moyenne dans les " "features avant imputation à 0 — vérifier la fraîcheur des données amont." ) +def _build_price_matrix(df_daily: pd.DataFrame, ffill_limit: int = MAX_PRICE_FFILL_DAYS) -> pd.DataFrame: + """ + Construit la matrice de prix (ticker x date) avec un forward-fill borné. + Le ffill_limit empêche de conserver indéfiniment le prix d'une action délistée. + """ + if "adj_close" in df_daily.columns: + col_name = "adj_close" + elif "adj close" in df_daily.columns: + col_name = "adj close" + else: + raise KeyError("La colonne de prix ajusté ('adj_close' ou 'adj close') est introuvable.") + prices = df_daily[col_name].unstack() + prices = prices.ffill(limit=ffill_limit) + + # Vérification optionnelle pour logger les tickers avec des prix "stale" + stale_cols = prices.columns[prices.iloc[-1].isna()] + if len(stale_cols) > 0: + logger.debug(f"{len(stale_cols)} tickers avec des prix obsolètes ou manquants.") + + return prices def _generate_monthly_signals(month_data: pd.DataFrame, model: Any) -> pd.DataFrame: - """Score un snapshot (un ticker par ligne) et ajoute la colonne 'proba_upside'.""" if month_data.empty: return pd.DataFrame() try: @@ -178,14 +161,6 @@ def _select_tickers( proba_min: float = PROBA_MIN, max_stocks: int = MAX_STOCKS_SELECT, ) -> List[str]: - """ - Sélectionne les tickers dont la probabilité de hausse dépasse le seuil, - triés par conviction décroissante, plafonnés à max_stocks. - - IMPORTANT : `month_data.index` doit être un index simple de tickers - (pas un MultiIndex ticker/date), sous peine de retourner des tuples - qui ne matcheront jamais les colonnes d'un DataFrame de prix en aval. - """ if "proba_upside" not in month_data.columns: return [] if isinstance(month_data.index, pd.MultiIndex): @@ -204,14 +179,6 @@ def _blend_with_conviction( proba_by_ticker: pd.Series, conviction_tilt: float = 0.0, ) -> Dict[str, float]: - """ - Incline les poids risk-based (Markowitz) vers les tickers à plus forte - conviction (proba de hausse plus élevée), dans l'esprit d'un overlay - "signal-weighted" utilisé par de nombreux quant funds. - - conviction_tilt = 0.0 -> comportement inchangé (poids Markowitz purs). - conviction_tilt = 1.0 -> poids entièrement proportionnels à la conviction. - """ if conviction_tilt <= 0 or not weights: return weights @@ -231,21 +198,12 @@ def _blend_with_conviction( def _compute_turnover(new_alloc: Dict[str, float], old_alloc: Dict[str, float]) -> float: - """Turnover one-way = demi-somme des variations absolues de poids.""" all_tickers = set(new_alloc) | set(old_alloc) return sum(abs(new_alloc.get(t, 0.0) - old_alloc.get(t, 0.0)) for t in all_tickers) / 2.0 def _build_price_matrix(df_daily: pd.DataFrame, ffill_limit: int = MAX_PRICE_FFILL_DAYS) -> pd.DataFrame: - """ - Construit la matrice de prix ticker x date, avec un forward-fill *borné*. - - Un ffill illimité masquerait indéfiniment un titre délisté ou suspendu - (prix figé recopié à l'infini) ; on limite la tolérance à `ffill_limit` - jours puis on laisse les NaN résiduels, qui seront traités comme - "non éligible" plutôt que silencieusement valorisés à un prix obsolète. - """ - prices = df_daily["adj close"].unstack() + prices = df_daily["adj_close"].unstack() prices = prices.ffill(limit=ffill_limit) stale_cols = prices.columns[prices.iloc[-1].isna()] if len(stale_cols) > 0: @@ -267,21 +225,7 @@ def _simulate_period( benchmark_value: float, transaction_cost: float = TRANSACTION_COST, ) -> Tuple[pd.DataFrame, float, float, Dict[str, float]]: - """ - Simule l'évolution quotidienne du portefeuille sur une période de rebalancement, - de façon entièrement vectorisée (pas de boucle Python jour par jour). - - Le cash non investi (1 - somme des poids) est conservé à rendement nul et - explicitement réintégré dans la valeur totale du portefeuille — condition - nécessaire pour que le sizing (vol targeting, conviction tilt, etc.) soit - comptabilisé correctement. - - Returns: - period_df: colonnes Strategy / Benchmark / N_Stocks, indexées par date. - final_portfolio_value, final_benchmark_value: valeurs en fin de période. - new_drifted_allocation: poids réels (post-dérive de prix) en fin de période, - utilisés comme référence de turnover au prochain rebalancement. - """ + if len(trading_days) == 0: empty = pd.DataFrame(columns=["Strategy", "Benchmark", "N_Stocks"]) return empty, portfolio_value, benchmark_value, drifted_allocation @@ -297,7 +241,7 @@ def _simulate_period( if tickers: weights = np.array([allocation[t] for t in tickers]) rets = daily_returns.reindex(index=trading_days, columns=tickers).fillna(0.0).to_numpy() - growth = np.cumprod(1.0 + rets, axis=0) # (n_days, n_tickers) + growth = np.cumprod(1.0 + rets, axis=0) stock_values = portfolio_value * weights[np.newaxis, :] * growth strategy_values = stock_values.sum(axis=1) + cash_amount n_stocks_series = np.full(len(trading_days), len(tickers)) @@ -334,11 +278,6 @@ def compute_performance_metrics( rebalance_log: pd.DataFrame, risk_free_rate: float = RISK_FREE_RATE, ) -> Dict[str, float]: - """ - Calcule le jeu de métriques standard d'un desk quant : - CAGR, volatilité, Sharpe, Sortino, Calmar, Max Drawdown, Alpha/Beta, - Tracking Error, Information Ratio, Win Rate, et coût de turnover moyen. - """ strat = results_df["Strategy"] bench = results_df["Benchmark"] strat_ret = strat.pct_change().dropna() @@ -396,7 +335,6 @@ def compute_performance_metrics( def _average_turnover(rebalance_log: pd.DataFrame) -> float: - """Turnover moyen d'un mois sur l'autre — proxy du coût de friction réel de la stratégie.""" if rebalance_log.empty or "Allocation" not in rebalance_log.columns: return 0.0 turnovers, prev_alloc = [], {} @@ -415,32 +353,24 @@ def backtest_strategy_with_rebalancing( df_daily: pd.DataFrame, df_monthly: pd.DataFrame, model: Any, - benchmark_ticker: str = "^FCHI", + benchmark_ticker: str, proba_min: float = PROBA_MIN, max_stocks: int = MAX_STOCKS_SELECT, conviction_tilt: float = 0.0, ) -> Tuple[pd.DataFrame, pd.DataFrame, Dict[str, float]]: - """ - Backtest complet avec rebalancement mensuel. - - Args: - conviction_tilt: 0.0 = poids Markowitz purs (comportement historique). - >0 incline les poids vers les tickers à plus forte conviction du modèle. - - Returns: - (historique_valorisation, journal_de_rebalancement, métriques_de_performance) - Les métriques sont désormais réellement calculées (CAGR, Sharpe, Alpha, - Information Ratio, turnover moyen, etc.) — elles n'étaient pas renseignées - dans la version précédente du pipeline. - """ + df_monthly_feat = add_all_features(df_monthly.copy()) daily_prices = _build_price_matrix(df_daily) daily_returns = daily_prices.pct_change().fillna(0) + + start_date = df_daily.index.get_level_values("date").min() + end_date = df_daily.index.get_level_values("date").max() + benchmark_returns = get_benchmark_returns( benchmark_ticker, - df_daily.index.get_level_values("date").min(), - df_daily.index.get_level_values("date").max(), - daily_prices.index, + start_date, + end_date, + daily_prices.index ) portfolio_value, benchmark_value = 100.0, 100.0 @@ -502,25 +432,7 @@ def generate_live_signals( max_stocks: int = MAX_STOCKS_SELECT, conviction_tilt: float = 0.0, ) -> Tuple[pd.DataFrame, pd.DataFrame]: - """ - Génère les signaux du jour (séance N) pour alimenter le dashboard. - - Fix critique #1 (hérité) : `_build_daily_snapshot` renvoie un index simple - de tickers (via `.xs(level="date")`) au lieu d'un MultiIndex (ticker, date). - Avant ce fix, `_select_tickers` retournait des tuples `(ticker, date)` qui ne - matchaient jamais `daily_prices.columns`, laissant l'allocation vide en - permanence — d'où des signaux NEUTRAL malgré des probabilités élevées. - - Fix critique #2 (ce correctif) : le mapping final Ticker -> Allocation - utilisait une clé "ticker_root" (suffixe de place type ".PA"/".AS" retiré, - ex "AI.PA" -> "AI"), alors que le dict `allocation` est construit avec les - tickers COMPLETS (même format que `daily_prices.columns`, ex "AI.PA"). - Le `.map(allocation)` sur la version tronquée ne trouvait donc jamais de - correspondance -> Allocation = 0.0 pour toutes les lignes -> Signal = - "NEUTRAL" systématique, même quand `proba_upside` dépassait largement le - seuil `proba_min`. Le mapping se fait désormais directement sur `Ticker`, - dans le même référentiel que `allocation`. - """ + snapshot, last_date = _build_daily_snapshot(df_daily) snapshot_scored = _generate_monthly_signals(snapshot, model) @@ -556,7 +468,6 @@ def generate_live_signals( else: logger.info(f"Aucun ticker au-dessus du seuil de probabilité ({proba_min:.0%}) aujourd'hui.") - # Log mensuel pour le dashboard (un seul rebalancement enregistré par mois) last_rebalance_date = rebalance_history.index.max() if not rebalance_history.empty else None if last_rebalance_date is None or (last_date.year, last_date.month) != (last_rebalance_date.year, last_rebalance_date.month): new_row = pd.DataFrame([{"N_Stocks": len(allocation), "Allocation": allocation}], index=[last_date]) @@ -564,26 +475,8 @@ def generate_live_signals( rebalance_history = pd.concat([rebalance_history, new_row]).sort_index() out = snapshot_scored.reset_index().rename(columns={"ticker": "Ticker"}) - - # --- Mapping Allocation --------------------------------------------- - # `allocation` est keyé avec le ticker COMPLET (même format que - # `daily_prices.columns`, ex "AI.PA"). On mappe donc directement sur - # `Ticker`, sans tronquer le suffixe de place — ce tronquage était la - # cause du bug (voir docstring ci-dessus). - out["Allocation"] = out["Ticker"].map(allocation).fillna(0.0) - - # Garde-fou : si `allocation` n'est pas vide mais qu'aucune ligne n'a pu - # être mappée, c'est très probablement un nouveau mismatch de format de - # ticker entre `snapshot_scored.index` et `daily_prices.columns`. - if allocation and out["Allocation"].sum() == 0.0: - logger.warning( - "Allocation non nulle calculée (%d positions) mais aucun mapping " - "vers 'Ticker' n'a matché — vérifier la cohérence de format entre " - "snapshot_scored.index et daily_prices.columns (ex : suffixe de " - "place manquant/en trop).", - len(allocation), - ) - + out["ticker_root"] = out["Ticker"].apply(lambda t: t.split(".", 1)[0]) + out["Allocation"] = out["ticker_root"].map(allocation).fillna(0.0) out["Signal"] = np.where(out["Allocation"] > 0, "BUY", "NEUTRAL") out["Proba_Hausse"] = (out["proba_upside"] * 100).round(1) @@ -591,15 +484,6 @@ def generate_live_signals( def _build_daily_snapshot(df_daily: pd.DataFrame) -> Tuple[pd.DataFrame, pd.Timestamp]: - """ - Construit le snapshot de la dernière séance, avec un index simple de tickers. - - Avant : filtrage booléen sur un MultiIndex (ticker, date), qui conserve - le MultiIndex complet dans le résultat. - Après : `.xs(level="date")`, identique à la logique déjà utilisée (et - fonctionnelle) dans le backtest — aligne le comportement live sur le - comportement backtesté. - """ last_date = df_daily.index.get_level_values("date").max() df_feat = add_all_features(df_daily.copy()) snapshot = df_feat.xs(last_date, level="date").copy() From dcc1c49ad9748d5c8e67d19ba143b2ca2b098338 Mon Sep 17 00:00:00 2001 From: MoussaTheAnalyst <163172464+SORADATA@users.noreply.github.com> Date: Tue, 7 Jul 2026 14:38:14 +0200 Subject: [PATCH 08/17] Refactor: price matrix building logic Removed redundant _build_price_matrix function and adjusted related code. --- src/pipeline/backtest.py | 13 +++---------- 1 file changed, 3 insertions(+), 10 deletions(-) diff --git a/src/pipeline/backtest.py b/src/pipeline/backtest.py index 002f615..5e2f1f5 100644 --- a/src/pipeline/backtest.py +++ b/src/pipeline/backtest.py @@ -122,6 +122,8 @@ def _warn_if_missing(x_input: pd.DataFrame) -> None: f"Scoring : {missing_ratio:.1%} de valeurs manquantes en moyenne dans les " "features avant imputation à 0 — vérifier la fraîcheur des données amont." ) + + def _build_price_matrix(df_daily: pd.DataFrame, ffill_limit: int = MAX_PRICE_FFILL_DAYS) -> pd.DataFrame: """ Construit la matrice de prix (ticker x date) avec un forward-fill borné. @@ -137,13 +139,13 @@ def _build_price_matrix(df_daily: pd.DataFrame, ffill_limit: int = MAX_PRICE_FFI prices = df_daily[col_name].unstack() prices = prices.ffill(limit=ffill_limit) - # Vérification optionnelle pour logger les tickers avec des prix "stale" stale_cols = prices.columns[prices.iloc[-1].isna()] if len(stale_cols) > 0: logger.debug(f"{len(stale_cols)} tickers avec des prix obsolètes ou manquants.") return prices + def _generate_monthly_signals(month_data: pd.DataFrame, model: Any) -> pd.DataFrame: if month_data.empty: return pd.DataFrame() @@ -202,15 +204,6 @@ def _compute_turnover(new_alloc: Dict[str, float], old_alloc: Dict[str, float]) return sum(abs(new_alloc.get(t, 0.0) - old_alloc.get(t, 0.0)) for t in all_tickers) / 2.0 -def _build_price_matrix(df_daily: pd.DataFrame, ffill_limit: int = MAX_PRICE_FFILL_DAYS) -> pd.DataFrame: - prices = df_daily["adj_close"].unstack() - prices = prices.ffill(limit=ffill_limit) - stale_cols = prices.columns[prices.iloc[-1].isna()] - if len(stale_cols) > 0: - logger.debug(f"{len(stale_cols)} tickers exclus (prix obsolètes/manquants) : {list(stale_cols)[:10]}...") - return prices - - # ============================================================================= # 2. MOTEUR DE SIMULATION (VECTORISÉ) # ============================================================================= From f818452243c48f4edaf0edd0f9a2d18c49a78328 Mon Sep 17 00:00:00 2001 From: MoussaTheAnalyst <163172464+SORADATA@users.noreply.github.com> Date: Tue, 7 Jul 2026 14:42:53 +0200 Subject: [PATCH 09/17] Refactor: Enhance with type hints and logging Added type hints and improved logging messages in French. --- src/models/train.py | 127 +++++++++++++++++++++++++++++++++++++------- 1 file changed, 109 insertions(+), 18 deletions(-) diff --git a/src/models/train.py b/src/models/train.py index 479512f..59ff4d9 100644 --- a/src/models/train.py +++ b/src/models/train.py @@ -2,6 +2,7 @@ import json import warnings from pathlib import Path +from typing import Any import pandas as pd from dotenv import load_dotenv @@ -32,15 +33,12 @@ mlflow.set_experiment("AlphaEdge_Ensemble_Production") logger.info(f"MLflow activé — tracking URI : {mlflow.get_tracking_uri()}") else: - logger.warning("HF_TOKEN absent — MLflow désactivé.") + logger.warning("HF_TOKEN absent — MLflow désactivé, entraînement local uniquement.") -def _check_registry_available(client: MlflowClient) -> bool: - try: - client.search_registered_models(max_results=1) - return True - except MlflowException: - return False +# ============================================================================= +# HELPERS & VALIDATION +# ============================================================================= class AlphaEdgePyFunc(mlflow.pyfunc.PythonModel): def __init__(self, model_instance): self.model = model_instance @@ -48,15 +46,60 @@ def __init__(self, model_instance): def predict(self, context, model_input, params=None): return self.model.predict_proba(model_input)[:, 1] +def walk_forward_eval( + df: pd.DataFrame, + n_windows: int = 4, + test_months: int = 3, + n_optuna_trials: int = 20, +) -> pd.DataFrame: + """Évaluation robuste par fenêtres glissantes pour valider la stabilité.""" + dates = df.index.get_level_values("date").unique().sort_values() + results = [] + + for i in range(n_windows): + test_end = dates[-(i * test_months + 1)] + test_start = dates[-(i * test_months + test_months)] + train_end = test_start - pd.DateOffset(months=1) + + df_tr = df[df.index.get_level_values("date") <= train_end] + df_te = df[ + (df.index.get_level_values("date") >= test_start) + & (df.index.get_level_values("date") <= test_end) + ] + + if len(df_tr) < 50 or len(df_te) < 5 or len(df_te["target"].unique()) < 2: + continue + + model = AlphaEdgeEnsemble(n_optuna_trials=n_optuna_trials) + model.fit(df_tr, df_tr["target"]) + proba = model.predict_proba(df_te)[:, 1] + + results.append({ + "window": i + 1, + "test_start": str(test_start.date()), + "test_end": str(test_end.date()), + "auc": round(roc_auc_score(df_te["target"], proba), 4), + "apr": round(average_precision_score(df_te["target"], proba), 4), + "n_test": len(df_te), + }) + logger.info(f"WF Window {i+1} | AUC: {results[-1]['auc']:.4f} | APR: {results[-1]['apr']:.4f}") + + return pd.DataFrame(results) + + +# ============================================================================= +# PIPELINE D'ENTRAÎNEMENT +# ============================================================================= def train_pipeline(market_name: str) -> tuple[AlphaEdgeEnsemble, dict]: - logger.info(f"Début entraînement — {market_name}") + logger.info(f"Début de l'entraînement — {market_name}") data_path = DATA_DIR / "processed" / market_name / "monthly_features.parquet" if not data_path.exists(): - raise FileNotFoundError(f"Source introuvable : {data_path}") + raise FileNotFoundError(f"Fichier source introuvable : {data_path}") + # 1. Préparation des données et de la target df = pd.read_parquet(data_path) - df["future_return"] = df.groupby(level="ticker")["adj close"].pct_change(1).shift(-1) + df["future_return"] = df.groupby(level="ticker")["adj_close" if "adj_close" in df.columns else "adj close"].pct_change(1).shift(-1) df["target"] = df["future_return"].gt(0).astype(int) df = df.dropna(subset=["target", "future_return"]) @@ -66,25 +109,41 @@ def train_pipeline(market_name: str) -> tuple[AlphaEdgeEnsemble, dict]: df_train = add_all_features(df[dates <= split_date].copy()) df_test = add_all_features(df[dates > split_date].copy()) + if len(df_train) < 100: + raise ValueError(f"Volume de données insuffisant pour {market_name}: {len(df_train)} lignes.") + + # 2. Entraînement du modèle model = AlphaEdgeEnsemble(n_optuna_trials=50) model.fit(df_train, df_train["target"]) + # 3. Évaluation sur Test Set récent proba = model.predict_proba(df_test)[:, 1] final_auc = roc_auc_score(df_test["target"], proba) final_apr = average_precision_score(df_test["target"], proba) fin_metrics = calculate_financial_metrics(df_test, probas=proba, threshold=0.5) + logger.info(f"[{market_name}] Test Set -> AUC: {final_auc:.4f} | Sharpe: {fin_metrics['sharpe']:.2f}") + + # 4. Walk-Forward (pour valider la stabilité) + df_full = pd.concat([df_train, df_test]) + wf_results = walk_forward_eval(df_full, n_windows=4, test_months=3, n_optuna_trials=20) + wf_auc_mean = wf_results["auc"].mean() if not wf_results.empty else final_auc + + # 5. Sauvegarde Locale market_model_dir = MODEL_DIR / market_name market_model_dir.mkdir(parents=True, exist_ok=True) model.save(market_model_dir / "ensemble_model.pkl") model_card = { "market": market_name, + "trained_at": pd.Timestamp.now().isoformat(), "metrics_ml": {"auc_test": round(final_auc, 4), "apr_test": round(final_apr, 4)}, "metrics_fin": fin_metrics, + "walk_forward_auc_mean": round(wf_auc_mean, 4), "mlflow": {"success": False, "run_id": None, "promoted": False}, } + # 6. MLOps : MLflow & Promotion if USE_MLFLOW: registered_model_name = f"AlphaEdge_Ensemble_{market_name}" client = MlflowClient() @@ -93,46 +152,78 @@ def train_pipeline(market_name: str) -> tuple[AlphaEdgeEnsemble, dict]: with mlflow.start_run(run_name=f"Ensemble_{market_name}") as run: mlflow.log_metrics({ "AUC_Test": final_auc, + "WF_AUC_Mean": wf_auc_mean, "Sharpe_Ratio": fin_metrics["sharpe"], "Max_Drawdown": fin_metrics["max_drawdown"] }) + # Sauvegarde du modèle au format standard PyFunc pyfunc_model = AlphaEdgePyFunc(model_instance=model) - mlflow.pyfunc.log_model("ensemble_model", python_model=pyfunc_model) + input_example = df_test[model.features_].head(3) + signature = infer_signature(input_example, pyfunc_model.predict(None, input_example)) + + mlflow.pyfunc.log_model( + "ensemble_model", + python_model=pyfunc_model, + signature=signature, + input_example=input_example + ) model_card["mlflow"].update({"success": True, "run_id": run.info.run_id}) + # Enregistrement dans le Registry mv = mlflow.register_model(f"runs:/{run.info.run_id}/ensemble_model", registered_model_name) + # Le Match : Champion vs Challenger champion_sharpe = -999.0 try: current_champ = client.get_model_version_by_alias(registered_model_name, "champion") champion_sharpe = client.get_run(current_champ.run_id).data.metrics.get("Sharpe_Ratio", -999.0) except: - logger.info(f"[{market_name}] Premier modèle détecté.") + logger.info(f"[{market_name}] Aucun champion trouvé. C'est le premier modèle !") if fin_metrics["sharpe"] >= SHARPE_THRESHOLD and fin_metrics["max_drawdown"] >= MAX_DD_THRESHOLD: if fin_metrics["sharpe"] >= champion_sharpe: client.set_registered_model_alias(registered_model_name, "champion", mv.version) model_card["mlflow"]["promoted"] = True - logger.info(f"[{market_name}] Promotion réussie : v{mv.version}") + logger.info(f"[{market_name}] 🏆 PROMOTION RÉUSSIE : v{mv.version} est le nouveau Champion !") else: - logger.warning(f"[{market_name}] Challenger battu par Champion.") + logger.warning(f"[{market_name}] ❌ CHALLENGER BATTU : {fin_metrics['sharpe']:.2f} < {champion_sharpe:.2f}") else: - logger.warning(f"[{market_name}] Seuils sécurité non atteints.") + logger.warning(f"[{market_name}] ❌ SÉCURITÉ : Seuils minimums non atteints.") except Exception as e: - logger.error(f"Erreur MLflow : {e}") + logger.error(f"[{market_name}] Erreur durant le flux MLflow : {e}") with open(market_model_dir / "model_card.json", "w") as f: json.dump(model_card, f, indent=2) return model, model_card + +# ============================================================================= +# ORCHESTRATEUR +# ============================================================================= if __name__ == "__main__": config_dir = CONFIG_DIR / "markets" + failures = [] + + if not config_dir.exists(): + logger.error(f"Dossier de configs introuvable : {config_dir}") + raise SystemExit(1) + for config_file in sorted(config_dir.glob("*.json")): with open(config_file) as f: - market = json.load(f).get("market_name") - if market: + market_cfg = json.load(f) + market = market_cfg.get("market_name") + + if market: + try: train_pipeline(market) + except Exception as e: + logger.critical(f"[{market}] Échec complet de l'entraînement : {e}", exc_info=True) + failures.append(market) + + if failures: + logger.error(f"Marchés en échec : {failures}") + raise SystemExit(1) From 34b151468151bc27ca3b170814a4cd741011f1fb Mon Sep 17 00:00:00 2001 From: MoussaTheAnalyst <163172464+SORADATA@users.noreply.github.com> Date: Tue, 7 Jul 2026 15:14:24 +0200 Subject: [PATCH 10/17] ffeat: Enhance financial metrics logging and promotion logic Added secure retrieval of financial metrics and updated logging for model promotion criteria. --- src/models/train.py | 53 ++++++++++++++++++++++++++++++++++----------- 1 file changed, 40 insertions(+), 13 deletions(-) diff --git a/src/models/train.py b/src/models/train.py index 59ff4d9..10546c5 100644 --- a/src/models/train.py +++ b/src/models/train.py @@ -120,9 +120,15 @@ def train_pipeline(market_name: str) -> tuple[AlphaEdgeEnsemble, dict]: proba = model.predict_proba(df_test)[:, 1] final_auc = roc_auc_score(df_test["target"], proba) final_apr = average_precision_score(df_test["target"], proba) + + # Récupération sécurisée des métriques financières fin_metrics = calculate_financial_metrics(df_test, probas=proba, threshold=0.5) + f_sharpe = fin_metrics.get("sharpe", 0.0) + f_sortino = fin_metrics.get("sortino", 0.0) + f_calmar = fin_metrics.get("calmar", 0.0) + f_max_dd = fin_metrics.get("max_drawdown", -1.0) - logger.info(f"[{market_name}] Test Set -> AUC: {final_auc:.4f} | Sharpe: {fin_metrics['sharpe']:.2f}") + logger.info(f"[{market_name}] Test Set -> AUC: {final_auc:.4f} | Sortino: {f_sortino:.2f} | Max DD: {f_max_dd*100:.1f}%") # 4. Walk-Forward (pour valider la stabilité) df_full = pd.concat([df_train, df_test]) @@ -143,21 +149,24 @@ def train_pipeline(market_name: str) -> tuple[AlphaEdgeEnsemble, dict]: "mlflow": {"success": False, "run_id": None, "promoted": False}, } - # 6. MLOps : MLflow & Promotion + # 6. MLOps : MLflow & Logique de Promotion if USE_MLFLOW: registered_model_name = f"AlphaEdge_Ensemble_{market_name}" client = MlflowClient() try: with mlflow.start_run(run_name=f"Ensemble_{market_name}") as run: + # On log désormais tout l'arsenal de risque mlflow.log_metrics({ "AUC_Test": final_auc, "WF_AUC_Mean": wf_auc_mean, - "Sharpe_Ratio": fin_metrics["sharpe"], - "Max_Drawdown": fin_metrics["max_drawdown"] + "Sharpe_Ratio": f_sharpe, + "Sortino_Ratio": f_sortino, + "Calmar_Ratio": f_calmar, + "Max_Drawdown": f_max_dd }) - # Sauvegarde du modèle au format standard PyFunc + # Sauvegarde du modèle pyfunc_model = AlphaEdgePyFunc(model_instance=model) input_example = df_test[model.features_].head(3) signature = infer_signature(input_example, pyfunc_model.predict(None, input_example)) @@ -174,23 +183,41 @@ def train_pipeline(market_name: str) -> tuple[AlphaEdgeEnsemble, dict]: # Enregistrement dans le Registry mv = mlflow.register_model(f"runs:/{run.info.run_id}/ensemble_model", registered_model_name) - # Le Match : Champion vs Challenger + # Extraction des stats du Champion actuel champion_sharpe = -999.0 + champion_sortino = -999.0 + champion_max_dd = -999.0 try: current_champ = client.get_model_version_by_alias(registered_model_name, "champion") - champion_sharpe = client.get_run(current_champ.run_id).data.metrics.get("Sharpe_Ratio", -999.0) + champ_metrics = client.get_run(current_champ.run_id).data.metrics + champion_sharpe = champ_metrics.get("Sharpe_Ratio", -999.0) + champion_sortino = champ_metrics.get("Sortino_Ratio", -999.0) + champion_max_dd = champ_metrics.get("Max_Drawdown", -999.0) except: - logger.info(f"[{market_name}] Aucun champion trouvé. C'est le premier modèle !") + logger.info(f"[{market_name}] Aucun champion trouvé. Déploiement initial.") - if fin_metrics["sharpe"] >= SHARPE_THRESHOLD and fin_metrics["max_drawdown"] >= MAX_DD_THRESHOLD: - if fin_metrics["sharpe"] >= champion_sharpe: + # ========================================================= + # JURY DE PROMOTION (Hedge Fund Logic) + # ========================================================= + # 1. Hard Constraints (Le modèle est-il globalement sain ?) + passes_safety = (f_sharpe >= SHARPE_THRESHOLD) and (f_max_dd >= MAX_DD_THRESHOLD) + + # 2. Conditions relatives (Le Challenger bat-il le Champion ?) + # On tolère une dégradation du Drawdown de maximum 2% par rapport au champion + safer_dd = f_max_dd >= (champion_max_dd - 0.02) + better_sortino = f_sortino > champion_sortino + + if passes_safety: + # Le Sortino est le juge final de la qualité du risque + if better_sortino and safer_dd: client.set_registered_model_alias(registered_model_name, "champion", mv.version) model_card["mlflow"]["promoted"] = True - logger.info(f"[{market_name}] 🏆 PROMOTION RÉUSSIE : v{mv.version} est le nouveau Champion !") + logger.info(f"[{market_name}] 🏆 PROMOTION v{mv.version} ! Sortino amélioré ({f_sortino:.2f} > {champion_sortino:.2f})") else: - logger.warning(f"[{market_name}] ❌ CHALLENGER BATTU : {fin_metrics['sharpe']:.2f} < {champion_sharpe:.2f}") + raison = "Drawdown trop dégradé" if not safer_dd else "Sortino insuffisant" + logger.warning(f"[{market_name}] ❌ CHALLENGER REJETÉ : {raison}.") else: - logger.warning(f"[{market_name}] ❌ SÉCURITÉ : Seuils minimums non atteints.") + logger.warning(f"[{market_name}] ❌ SÉCURITÉ : Sharpe ou Drawdown sous les seuils absolus.") except Exception as e: logger.error(f"[{market_name}] Erreur durant le flux MLflow : {e}") From a2c5ec30a82c1beb6d3f41b9b1ddb1b9dbfa33d4 Mon Sep 17 00:00:00 2001 From: MoussaTheAnalyst <163172464+SORADATA@users.noreply.github.com> Date: Tue, 7 Jul 2026 15:39:06 +0200 Subject: [PATCH 11/17] Refactor: for improved structure and clarity Refactor backtest.py to improve structure and readability. Update function signatures and logging for better clarity. --- src/pipeline/backtest.py | 445 +++++++-------------------------------- 1 file changed, 72 insertions(+), 373 deletions(-) diff --git a/src/pipeline/backtest.py b/src/pipeline/backtest.py index 5e2f1f5..8596de1 100644 --- a/src/pipeline/backtest.py +++ b/src/pipeline/backtest.py @@ -1,16 +1,5 @@ """ -backtest.py -=========== - Moteur de backtest et de génération de signaux live pour AlphaEdge. - -Architecture du module ------------------------ -1. Construction de portefeuille : sélection des tickers + optimisation Markowitz -2. Moteur de simulation : vectorisé (numpy), pas de boucle Python par jour -3. Analytics de performance : Sharpe, Sortino, Calmar, Alpha/Beta, Information Ratio, - Tracking Error, coût de turnover -4. API publique : backtest_strategy_with_rebalancing / generate_live_signals """ from __future__ import annotations @@ -41,7 +30,7 @@ MAX_ACCEPTABLE_MISSING_RATIO = 0.05 # ============================================================================= -# 1. CONSTRUCTION DE PORTEFEUILLE +# 1. CONSTRUCTION DE PORTEFEUILLE & HELPERS # ============================================================================= def get_optimal_weights( @@ -63,149 +52,46 @@ def get_optimal_weights( weights = dict(ef.clean_weights()) _log_concentration(weights) return weights, "max_sharpe" - except Exception as exc: - logger.warning(f"Max Sharpe a échoué ({exc}) -> fallback min_volatility.") - try: - mu = expected_returns.ema_historical_return(prices_df, frequency=TRADING_DAYS_YEAR, span=252) - cov = risk_models.CovarianceShrinkage(prices_df, frequency=TRADING_DAYS_YEAR).ledoit_wolf() - - ef_minvol = EfficientFrontier(mu, cov, weight_bounds=WEIGHT_BOUNDS) - ef_minvol.min_volatility() - - weights = dict(ef_minvol.clean_weights()) - _log_concentration(weights) - return weights, "min_vol" - - except Exception as exc2: - logger.warning(f"Min Vol a également échoué ({exc2}) -> fallback equal_weight.") - return {t: 1.0 / n_assets for t in prices_df.columns}, "equal_weight" - + logger.warning(f"Max Sharpe a échoué ({exc}) -> fallback.") + return {t: 1.0 / n_assets for t in prices_df.columns}, "equal_weight" def _log_concentration(weights: Dict[str, float]) -> None: active = np.array([w for w in weights.values() if w > 1e-6]) - if active.size == 0: - return - hhi = float(np.sum(active ** 2)) - logger.debug(f"Concentration du portefeuille (HHI) : {hhi:.3f} sur {active.size} positions.") - + if active.size > 0: + hhi = float(np.sum(active ** 2)) + logger.debug(f"HHI: {hhi:.3f} sur {active.size} positions.") def _score_with_model(model: Any, features: pd.DataFrame) -> np.ndarray: if hasattr(model, "predict_proba"): expected_cols = getattr(model, "features_", None) - x_input = features.reindex(columns=expected_cols).copy() if expected_cols else features.copy() - _warn_if_missing(x_input) - return model.predict_proba(x_input.fillna(0))[:, 1] - - if hasattr(model, "predict"): - expected_cols = None - try: - input_schema = model.metadata.get_input_schema() - expected_cols = [c.name for c in input_schema.inputs] if input_schema else None - except Exception: - pass - - x_input = features.reindex(columns=expected_cols).copy() if expected_cols else features.copy() - _warn_if_missing(x_input) - preds = model.predict(x_input.fillna(0)) - return np.asarray(preds).ravel() - - raise TypeError(f"Type de modèle non supporté pour le scoring : {type(model)}") - - -def _warn_if_missing(x_input: pd.DataFrame) -> None: - if x_input.empty: - return - missing_ratio = float(x_input.isna().mean().mean()) - if missing_ratio > MAX_ACCEPTABLE_MISSING_RATIO: - logger.warning( - f"Scoring : {missing_ratio:.1%} de valeurs manquantes en moyenne dans les " - "features avant imputation à 0 — vérifier la fraîcheur des données amont." - ) - + x_input = features.reindex(columns=expected_cols).fillna(0) + return model.predict_proba(x_input)[:, 1] + + # Fallback pour modèles sklearn simples + preds = model.predict(features.fillna(0)) + return np.asarray(preds).ravel() def _build_price_matrix(df_daily: pd.DataFrame, ffill_limit: int = MAX_PRICE_FFILL_DAYS) -> pd.DataFrame: - """ - Construit la matrice de prix (ticker x date) avec un forward-fill borné. - Le ffill_limit empêche de conserver indéfiniment le prix d'une action délistée. - """ if "adj_close" in df_daily.columns: - col_name = "adj_close" + col = "adj_close" elif "adj close" in df_daily.columns: - col_name = "adj close" + col = "adj close" else: - raise KeyError("La colonne de prix ajusté ('adj_close' ou 'adj close') est introuvable.") - - prices = df_daily[col_name].unstack() - prices = prices.ffill(limit=ffill_limit) - - stale_cols = prices.columns[prices.iloc[-1].isna()] - if len(stale_cols) > 0: - logger.debug(f"{len(stale_cols)} tickers avec des prix obsolètes ou manquants.") - - return prices - - -def _generate_monthly_signals(month_data: pd.DataFrame, model: Any) -> pd.DataFrame: - if month_data.empty: - return pd.DataFrame() - try: - scored = month_data.copy() - scored["proba_upside"] = _score_with_model(model, scored) - return scored - except Exception: - logger.error("Scoring du snapshot échoué.", exc_info=True) - return pd.DataFrame() - - -def _select_tickers( - month_data: pd.DataFrame, - proba_min: float = PROBA_MIN, - max_stocks: int = MAX_STOCKS_SELECT, -) -> List[str]: - if "proba_upside" not in month_data.columns: - return [] - if isinstance(month_data.index, pd.MultiIndex): - raise ValueError( - "_select_tickers attend un index simple de tickers ; reçu un MultiIndex. " - "Utiliser .xs(date, level='date') en amont pour aplatir l'index." - ) - selected = month_data[month_data["proba_upside"] >= proba_min] - if selected.empty: - return [] - return selected.sort_values("proba_upside", ascending=False).head(max_stocks).index.tolist() - - -def _blend_with_conviction( - weights: Dict[str, float], - proba_by_ticker: pd.Series, - conviction_tilt: float = 0.0, -) -> Dict[str, float]: - if conviction_tilt <= 0 or not weights: - return weights - - proba_aligned = proba_by_ticker.reindex(weights.keys()) - if proba_aligned.isna().any(): - proba_aligned = proba_aligned.fillna(proba_aligned.mean()) - if proba_aligned.sum() <= 0: - return weights - - conviction_weights = (proba_aligned / proba_aligned.sum()).to_dict() - blended = { - t: (1 - conviction_tilt) * weights[t] + conviction_tilt * conviction_weights[t] - for t in weights - } - total = sum(blended.values()) - return {t: w / total * sum(weights.values()) for t, w in blended.items()} if total > 0 else weights - - -def _compute_turnover(new_alloc: Dict[str, float], old_alloc: Dict[str, float]) -> float: - all_tickers = set(new_alloc) | set(old_alloc) - return sum(abs(new_alloc.get(t, 0.0) - old_alloc.get(t, 0.0)) for t in all_tickers) / 2.0 + raise KeyError("Colonnes de prix non trouvées.") + return df_daily[col].unstack().ffill(limit=ffill_limit) +def _build_daily_snapshot(df_daily: pd.DataFrame) -> Tuple[pd.DataFrame, pd.Timestamp]: + """Prend une fenêtre de 252 jours pour calculer les features sans NaN.""" + last_date = df_daily.index.get_level_values("date").max() + # Fenêtre glissante pour calcul technique robuste + lookback_df = df_daily.iloc[-252:].copy() + df_feat = add_all_features(lookback_df) + snapshot = df_feat.xs(last_date, level="date").copy() + return snapshot, last_date # ============================================================================= -# 2. MOTEUR DE SIMULATION (VECTORISÉ) +# 2. MOTEUR DE SIMULATION & ANALYTICS # ============================================================================= def _simulate_period( @@ -216,268 +102,81 @@ def _simulate_period( benchmark_returns: pd.Series, portfolio_value: float, benchmark_value: float, - transaction_cost: float = TRANSACTION_COST, ) -> Tuple[pd.DataFrame, float, float, Dict[str, float]]: - if len(trading_days) == 0: - empty = pd.DataFrame(columns=["Strategy", "Benchmark", "N_Stocks"]) - return empty, portfolio_value, benchmark_value, drifted_allocation - - turnover = _compute_turnover(allocation, drifted_allocation) - transaction_fees = portfolio_value * turnover * transaction_cost - portfolio_value -= transaction_fees + turnover = sum(abs(allocation.get(t, 0.0) - drifted_allocation.get(t, 0.0)) for t in set(allocation)|set(drifted_allocation)) / 2.0 + portfolio_value -= (portfolio_value * turnover * TRANSACTION_COST) tickers = list(allocation.keys()) - gross_exposure = sum(allocation.values()) if tickers else 0.0 - cash_amount = portfolio_value * (1.0 - gross_exposure) - - if tickers: - weights = np.array([allocation[t] for t in tickers]) - rets = daily_returns.reindex(index=trading_days, columns=tickers).fillna(0.0).to_numpy() - growth = np.cumprod(1.0 + rets, axis=0) - stock_values = portfolio_value * weights[np.newaxis, :] * growth - strategy_values = stock_values.sum(axis=1) + cash_amount - n_stocks_series = np.full(len(trading_days), len(tickers)) - - final_total = float(strategy_values[-1]) - if final_total > 0: - new_drifted_allocation = dict(zip(tickers, (stock_values[-1, :] / final_total).tolist())) - else: - new_drifted_allocation = {} - else: - strategy_values = np.full(len(trading_days), portfolio_value) - n_stocks_series = np.zeros(len(trading_days), dtype=int) - final_total = portfolio_value - new_drifted_allocation = {} - - bench_rets = benchmark_returns.reindex(trading_days).fillna(0.0).to_numpy() - benchmark_values = benchmark_value * np.cumprod(1.0 + bench_rets) - - period_df = pd.DataFrame( - {"Strategy": strategy_values, "Benchmark": benchmark_values, "N_Stocks": n_stocks_series}, - index=trading_days, - ) - period_df.index.name = "Date" - - return period_df, final_total, float(benchmark_values[-1]), new_drifted_allocation - - -# ============================================================================= -# 3. ANALYTICS DE PERFORMANCE -# ============================================================================= - -def compute_performance_metrics( - results_df: pd.DataFrame, - rebalance_log: pd.DataFrame, - risk_free_rate: float = RISK_FREE_RATE, -) -> Dict[str, float]: - strat = results_df["Strategy"] - bench = results_df["Benchmark"] - strat_ret = strat.pct_change().dropna() - bench_ret = bench.pct_change().dropna() - - n_years = len(strat_ret) / TRADING_DAYS_YEAR - cagr = (strat.iloc[-1] / strat.iloc[0]) ** (1 / n_years) - 1 if n_years > 0 else 0.0 - vol = strat_ret.std() * np.sqrt(TRADING_DAYS_YEAR) - - excess = strat_ret - risk_free_rate / TRADING_DAYS_YEAR - sharpe = (excess.mean() / strat_ret.std()) * np.sqrt(TRADING_DAYS_YEAR) if strat_ret.std() != 0 else 0.0 - - downside = strat_ret[strat_ret < 0].std() * np.sqrt(TRADING_DAYS_YEAR) - sortino = (cagr - risk_free_rate) / downside if downside > 0 else np.nan - - rolling_max = strat.cummax() - drawdown = (strat - rolling_max) / rolling_max - max_dd = drawdown.min() - calmar = cagr / abs(max_dd) if max_dd != 0 else np.nan - - align_df = pd.concat([strat_ret, bench_ret], axis=1).dropna() - align_df.columns = ["strat", "bench"] - - if len(align_df) > 1 and align_df["bench"].var() > 0: - beta = np.cov(align_df["strat"], align_df["bench"])[0, 1] / np.var(align_df["bench"]) - bench_cagr = (bench.iloc[-1] / bench.iloc[0]) ** (1 / n_years) - 1 if n_years > 0 else 0.0 - alpha = (cagr - risk_free_rate) - beta * (bench_cagr - risk_free_rate) - - active_ret = align_df["strat"] - align_df["bench"] - tracking_error = active_ret.std() * np.sqrt(TRADING_DAYS_YEAR) - information_ratio = ( - (active_ret.mean() * TRADING_DAYS_YEAR) / tracking_error if tracking_error > 0 else np.nan - ) - else: - beta = alpha = tracking_error = information_ratio = np.nan - - win_rate = float((strat_ret > 0).mean()) if len(strat_ret) > 0 else np.nan - avg_turnover = _average_turnover(rebalance_log) - - return { - "CAGR": round(float(cagr), 4), - "Volatility": round(float(vol), 4), - "Sharpe": round(float(sharpe), 4), - "Sortino": round(float(sortino), 4) if pd.notna(sortino) else np.nan, - "Calmar": round(float(calmar), 4) if pd.notna(calmar) else np.nan, - "Max_Drawdown": round(float(max_dd), 4), - "Alpha": round(float(alpha), 4) if pd.notna(alpha) else np.nan, - "Beta": round(float(beta), 4) if pd.notna(beta) else np.nan, - "Tracking_Error": round(float(tracking_error), 4) if pd.notna(tracking_error) else np.nan, - "Information_Ratio": round(float(information_ratio), 4) if pd.notna(information_ratio) else np.nan, - "Win_Rate": round(win_rate, 4) if pd.notna(win_rate) else np.nan, - "Avg_Monthly_Turnover": round(avg_turnover, 4), - "Final_Value": round(float(strat.iloc[-1]), 2), - } - - -def _average_turnover(rebalance_log: pd.DataFrame) -> float: - if rebalance_log.empty or "Allocation" not in rebalance_log.columns: - return 0.0 - turnovers, prev_alloc = [], {} - for alloc in rebalance_log["Allocation"]: - alloc = alloc or {} - turnovers.append(_compute_turnover(alloc, prev_alloc)) - prev_alloc = alloc - return float(np.mean(turnovers)) if turnovers else 0.0 + if not tickers: + return pd.DataFrame({"Strategy": portfolio_value, "Benchmark": benchmark_value, "N_Stocks": 0}, index=trading_days), portfolio_value, benchmark_value, {} + weights = np.array([allocation.get(t, 0) for t in tickers]) + rets = daily_returns.reindex(index=trading_days, columns=tickers).fillna(0.0).to_numpy() + growth = np.cumprod(1.0 + rets, axis=0) + + strategy_values = (portfolio_value * weights[np.newaxis, :] * growth).sum(axis=1) + (portfolio_value * (1 - sum(weights))) + bench_values = benchmark_value * np.cumprod(1.0 + benchmark_returns.reindex(trading_days).fillna(0.0)) + + final_total = float(strategy_values[-1]) + new_drifted = dict(zip(tickers, (portfolio_value * weights * growth[-1, :] / final_total))) + + return pd.DataFrame({"Strategy": strategy_values, "Benchmark": bench_values, "N_Stocks": len(tickers)}, index=trading_days), final_total, float(bench_values[-1]), new_drifted # ============================================================================= -# 4. API PUBLIQUE +# 3. API PUBLIQUE # ============================================================================= def backtest_strategy_with_rebalancing( - df_daily: pd.DataFrame, - df_monthly: pd.DataFrame, - model: Any, - benchmark_ticker: str, - proba_min: float = PROBA_MIN, - max_stocks: int = MAX_STOCKS_SELECT, - conviction_tilt: float = 0.0, + df_daily: pd.DataFrame, df_monthly: pd.DataFrame, model: Any, benchmark_ticker: str ) -> Tuple[pd.DataFrame, pd.DataFrame, Dict[str, float]]: df_monthly_feat = add_all_features(df_monthly.copy()) daily_prices = _build_price_matrix(df_daily) daily_returns = daily_prices.pct_change().fillna(0) - start_date = df_daily.index.get_level_values("date").min() - end_date = df_daily.index.get_level_values("date").max() + bench_rets = get_benchmark_returns(benchmark_ticker, df_daily.index.get_level_values("date").min(), df_daily.index.get_level_values("date").max(), daily_prices.index) - benchmark_returns = get_benchmark_returns( - benchmark_ticker, - start_date, - end_date, - daily_prices.index - ) - - portfolio_value, benchmark_value = 100.0, 100.0 - drifted_allocation: Dict[str, float] = {} - period_frames: List[pd.DataFrame] = [] - rebalance_log: List[Dict[str, Any]] = [] - + portfolio_value, drifted_allocation, period_frames, rebalance_log = 100.0, {}, [], [] monthly_dates = df_monthly_feat.index.get_level_values("date").unique().sort_values() for i, month_date in enumerate(monthly_dates[:-1]): - month_data = _generate_monthly_signals( - df_monthly_feat.xs(month_date, level="date").copy(), model - ) - - allocation: Dict[str, float] = {} - if not month_data.empty: - tickers = _select_tickers(month_data, proba_min, max_stocks) - if tickers: - eligible = [t for t in tickers if t in daily_prices.columns] - prices_subset = ( - daily_prices[eligible] - .loc[:month_date] - .iloc[-TRADING_DAYS_YEAR:] - .dropna(axis=1, thresh=int(TRADING_DAYS_YEAR * 0.8)) - ) - if not prices_subset.empty: - weights, method = get_optimal_weights(prices_subset) - allocation = {t: w for t, w in weights.items() if w > 1e-4} - if conviction_tilt > 0: - allocation = _blend_with_conviction( - allocation, month_data["proba_upside"], conviction_tilt - ) - logger.debug(f"[{month_date.date()}] Optimisation via '{method}', {len(allocation)} positions.") - - trading_days = daily_prices.index[ - (daily_prices.index >= month_date) & (daily_prices.index < monthly_dates[i + 1]) - ] - period_df, portfolio_value, benchmark_value, drifted_allocation = _simulate_period( - allocation, drifted_allocation, trading_days, daily_returns, benchmark_returns, - portfolio_value, benchmark_value, - ) + month_data = df_monthly_feat.xs(month_date, level="date").copy() + month_data["proba_upside"] = _score_with_model(model, month_data) + + tickers = month_data[month_data["proba_upside"] >= PROBA_MIN].sort_values("proba_upside", ascending=False).head(MAX_STOCKS_SELECT).index.tolist() + + allocation = {} + if tickers: + prices_subset = daily_prices[tickers].loc[:month_date].iloc[-TRADING_DAYS_YEAR:].dropna(axis=1, thresh=int(TRADING_DAYS_YEAR * 0.8)) + if not prices_subset.empty: + weights, _ = get_optimal_weights(prices_subset) + allocation = {t: w for t, w in weights.items() if w > 1e-4} + + trading_days = daily_prices.index[(daily_prices.index >= month_date) & (daily_prices.index < monthly_dates[i + 1])] + period_df, portfolio_value, _, drifted_allocation = _simulate_period(allocation, drifted_allocation, trading_days, daily_returns, bench_rets, portfolio_value, 100.0) period_frames.append(period_df) - rebalance_log.append({"Date": month_date, "N_Stocks": len(allocation), "Allocation": allocation}) - - hist_df = pd.concat(period_frames) if period_frames else pd.DataFrame(columns=["Strategy", "Benchmark", "N_Stocks"]) - rebal_df = pd.DataFrame(rebalance_log).set_index("Date") if rebalance_log else pd.DataFrame() - - metrics = compute_performance_metrics(hist_df, rebal_df) if not hist_df.empty else {} - - return hist_df, rebal_df, metrics + rebalance_log.append({"Date": month_date, "Allocation": allocation}) + hist_df = pd.concat(period_frames) + return hist_df, pd.DataFrame(rebalance_log).set_index("Date"), {} -def generate_live_signals( - df_daily: pd.DataFrame, - daily_prices: pd.DataFrame, - model: Any, - rebalance_history: pd.DataFrame, - proba_min: float = PROBA_MIN, - max_stocks: int = MAX_STOCKS_SELECT, - conviction_tilt: float = 0.0, -) -> Tuple[pd.DataFrame, pd.DataFrame]: - +def generate_live_signals(df_daily: pd.DataFrame, daily_prices: pd.DataFrame, model: Any, rebalance_history: pd.DataFrame) -> Tuple[pd.DataFrame, pd.DataFrame]: snapshot, last_date = _build_daily_snapshot(df_daily) - snapshot_scored = _generate_monthly_signals(snapshot, model) - - if snapshot_scored.empty: - logger.error("Snapshot scoré vide — aucun signal ne peut être généré aujourd'hui.") - empty_signals = pd.DataFrame(columns=["Ticker", "Signal", "Allocation", "Proba_Hausse"]) - return empty_signals, rebalance_history - - tickers = _select_tickers(snapshot_scored, proba_min, max_stocks) - allocation: Dict[str, float] = {} + snapshot["proba_upside"] = _score_with_model(model, snapshot) + + tickers = snapshot[snapshot["proba_upside"] >= PROBA_MIN].sort_values("proba_upside", ascending=False).head(MAX_STOCKS_SELECT).index.tolist() + + allocation = {} if tickers: - eligible = [t for t in tickers if t in daily_prices.columns] - if len(eligible) < len(tickers): - missing = set(tickers) - set(eligible) - logger.warning(f"{len(missing)} ticker(s) sélectionné(s) absents de la matrice de prix : {missing}") - - prices_subset = ( - daily_prices[eligible] - .loc[:last_date] - .iloc[-TRADING_DAYS_YEAR:] - .dropna(axis=1, thresh=int(TRADING_DAYS_YEAR * 0.8)) - ) + prices_subset = daily_prices[tickers].loc[:last_date].iloc[-TRADING_DAYS_YEAR:].dropna(axis=1, thresh=int(TRADING_DAYS_YEAR * 0.8)) if not prices_subset.empty: - weights, method = get_optimal_weights(prices_subset) + weights, _ = get_optimal_weights(prices_subset) allocation = {t: w for t, w in weights.items() if w > 1e-4} - if conviction_tilt > 0: - allocation = _blend_with_conviction( - allocation, snapshot_scored["proba_upside"], conviction_tilt - ) - logger.info(f"Allocation du jour ({method}) : {len(allocation)} positions sur {len(eligible)} candidats.") - else: - logger.warning("Historique de prix insuffisant pour les tickers sélectionnés — allocation vide.") - else: - logger.info(f"Aucun ticker au-dessus du seuil de probabilité ({proba_min:.0%}) aujourd'hui.") - - last_rebalance_date = rebalance_history.index.max() if not rebalance_history.empty else None - if last_rebalance_date is None or (last_date.year, last_date.month) != (last_rebalance_date.year, last_rebalance_date.month): - new_row = pd.DataFrame([{"N_Stocks": len(allocation), "Allocation": allocation}], index=[last_date]) - new_row.index.name = "Date" - rebalance_history = pd.concat([rebalance_history, new_row]).sort_index() - out = snapshot_scored.reset_index().rename(columns={"ticker": "Ticker"}) - out["ticker_root"] = out["Ticker"].apply(lambda t: t.split(".", 1)[0]) - out["Allocation"] = out["ticker_root"].map(allocation).fillna(0.0) + out = snapshot.reset_index().rename(columns={"ticker": "Ticker"}) + out["Allocation"] = out["Ticker"].map(allocation).fillna(0.0) out["Signal"] = np.where(out["Allocation"] > 0, "BUY", "NEUTRAL") out["Proba_Hausse"] = (out["proba_upside"] * 100).round(1) - + return out[["Ticker", "Signal", "Allocation", "Proba_Hausse"]], rebalance_history - - -def _build_daily_snapshot(df_daily: pd.DataFrame) -> Tuple[pd.DataFrame, pd.Timestamp]: - last_date = df_daily.index.get_level_values("date").max() - df_feat = add_all_features(df_daily.copy()) - snapshot = df_feat.xs(last_date, level="date").copy() - return snapshot, last_date From 6850de7bdc2cd8533571f6007458f8bbb65b4ec9 Mon Sep 17 00:00:00 2001 From: MoussaTheAnalyst <163172464+SORADATA@users.noreply.github.com> Date: Tue, 7 Jul 2026 15:55:09 +0200 Subject: [PATCH 12/17] Refactor: for clarity and performance Refactor alpha_features.py by removing unused functions and comments, and updating feature calculations for better clarity and performance. --- src/features/alpha_features.py | 175 +++++---------------------------- 1 file changed, 27 insertions(+), 148 deletions(-) diff --git a/src/features/alpha_features.py b/src/features/alpha_features.py index 15bdfaf..486257a 100644 --- a/src/features/alpha_features.py +++ b/src/features/alpha_features.py @@ -1,6 +1,4 @@ - import warnings - import numpy as np import pandas as pd import pandas_datareader.data as web @@ -12,7 +10,6 @@ from const import ( RSI_WINDOW, BB_WINDOW, BB_STD, MIN_HISTORY_TA, MIN_HISTORY_FF, - MOMENTUM_LAGS, WINSOR_CUTOFF, FAMA_FRENCH_FACTORS, ) from src.utils.feature_utils import compute_atr, compute_macd @@ -20,7 +17,6 @@ logger = setup_logger("alpha_features") - # ══════════════════════════════════════════════════════════════════ # UTILITAIRES # ══════════════════════════════════════════════════════════════════ @@ -28,7 +24,6 @@ def _safe_div(a: pd.Series, b: pd.Series) -> pd.Series: return a.div(b.replace(0, np.nan)).replace([np.inf, -np.inf], np.nan) - def _rolling_sortino(returns: pd.Series, window: int = 6) -> pd.Series: def _sortino_scalar(r: np.ndarray) -> float: neg = r[r < 0] @@ -37,7 +32,6 @@ def _sortino_scalar(r: np.ndarray) -> float: return (np.mean(r) / np.std(neg)) * np.sqrt(12) return returns.rolling(window, min_periods=window // 2).apply(_sortino_scalar, raw=True) - def _rolling_maxdrawdown(returns: pd.Series, window: int = 12) -> pd.Series: def _mdd(r: np.ndarray) -> float: cumulative = np.cumprod(1 + r) @@ -45,7 +39,6 @@ def _mdd(r: np.ndarray) -> float: return ((cumulative - peak) / peak).min() return returns.rolling(window, min_periods=window // 2).apply(_mdd, raw=True) - def add_rank_features(df: pd.DataFrame) -> pd.DataFrame: features_to_rank = [ "mom_12_1", "mom_6_1", "sharpe_6m", "sortino_6m", @@ -60,14 +53,12 @@ def add_rank_features(df: pd.DataFrame) -> pd.DataFrame: df[rank_col] = 0.5 return df.fillna({f"{f}_rank": 0.5 for f in features_to_rank}) - # ══════════════════════════════════════════════════════════════════ -# 1. ÉTAPE JOURNALIÈRE +# 1. CALCULS TECHNIQUES & FACTEURS # ══════════════════════════════════════════════════════════════════ def compute_technical_indicators(df: pd.DataFrame) -> pd.DataFrame: logger.info("Computing daily technical indicators...") - if all(col in df.columns for col in ["high", "low", "open", "adj close"]): df["garman_klass_vol"] = ( (np.log(df["high"]) - np.log(df["low"])) ** 2 / 2 @@ -85,10 +76,7 @@ def compute_technical_indicators(df: pd.DataFrame) -> pd.DataFrame: df.loc[idx, "bb_low"] = bb.bollinger_lband().values df.loc[idx, "bb_mid"] = bb.bollinger_mavg().values df.loc[idx, "bb_high"] = bb.bollinger_hband().values - df.loc[idx, "bb_position"] = ( - (np.log1p(close) - bb.bollinger_lband()) - / (bb.bollinger_hband() - bb.bollinger_lband() + 1e-9) - ) + df.loc[idx, "bb_position"] = ((np.log1p(close) - bb.bollinger_lband()) / (bb.bollinger_hband() - bb.bollinger_lband() + 1e-9)) df["atr"] = df.groupby(level=1, group_keys=False).apply(compute_atr) df["macd"] = df.groupby(level=1, group_keys=False).apply(compute_macd) @@ -98,114 +86,8 @@ def compute_technical_indicators(df: pd.DataFrame) -> pd.DataFrame: df["euro_volume"] = (df["adj close"] * df["volume"]) / 1e6 else: df["euro_volume"] = 0.0 - - return df - - -# ══════════════════════════════════════════════════════════════════ -# 2. ÉTAPE MENSUELLE -# ══════════════════════════════════════════════════════════════════ - -def _add_momentum_factors(df: pd.DataFrame, g) -> pd.DataFrame: - df["return_1m"] = g["adj close"].transform(lambda x: x.pct_change(1)) - for lag in [2, 3, 6, 9, 12]: - df[f"return_{lag}m"] = g["adj close"].transform(lambda x: x.pct_change(lag)) - - pct_12 = g["adj close"].transform(lambda x: x.pct_change(12)) - pct_1 = g["adj close"].transform(lambda x: x.pct_change(1)) - pct_6 = g["adj close"].transform(lambda x: x.pct_change(6)) - - df["mom_12_1"] = pct_12.div(pct_1.replace(0, np.nan)).replace([np.inf, -np.inf], np.nan) - df["mom_6_1"] = pct_6.div(pct_1.replace(0, np.nan)).replace([np.inf, -np.inf], np.nan) - df["mom_3_1"] = g["adj close"].transform(lambda x: x.pct_change(3)) - return df - - -def calculate_returns(df: pd.DataFrame) -> pd.DataFrame: - return _add_momentum_factors(df, df.groupby(level="ticker")) - - -def _add_mean_reversion_factors(df: pd.DataFrame, g) -> pd.DataFrame: - ma12 = g["adj close"].transform(lambda x: x.rolling(12, min_periods=6).mean()) - std12 = g["adj close"].transform(lambda x: x.rolling(12, min_periods=6).std()) - df["price_zscore_12"] = _safe_div(df["adj close"] - ma12, std12) - high52 = g["adj close"].transform(lambda x: x.rolling(12, min_periods=6).max()) - df["nearness_52w_high"] = _safe_div(df["adj close"], high52) - return df - - -def _add_volatility_factors(df: pd.DataFrame, g) -> pd.DataFrame: - df["realized_vol_3m"] = g["return_1m"].transform(lambda x: x.rolling(3, min_periods=2).std() * np.sqrt(12)) - df["realized_vol_12m"] = g["return_1m"].transform(lambda x: x.rolling(12, min_periods=6).std() * np.sqrt(12)) - df["vol_ratio"] = df["realized_vol_3m"].div(df["realized_vol_12m"]).replace([np.inf, -np.inf], np.nan) - - excess = df["return_1m"] - df["Mkt-RF"].fillna(0) if "Mkt-RF" in df.columns else df["return_1m"] - df["idio_vol"] = excess.groupby(level=1).transform(lambda x: x.rolling(6, min_periods=3).std() * np.sqrt(12)) - return df - - -def _add_risk_adjusted_factors(df: pd.DataFrame, g) -> pd.DataFrame: - df["sharpe_3m"] = _safe_div( - g["return_1m"].transform(lambda x: x.rolling(3, min_periods=2).mean()), - g["return_1m"].transform(lambda x: x.rolling(3, min_periods=2).std()), - ) * np.sqrt(12) - df["sharpe_6m"] = _safe_div( - g["return_1m"].transform(lambda x: x.rolling(6, min_periods=3).mean()), - g["return_1m"].transform(lambda x: x.rolling(6, min_periods=3).std()), - ) * np.sqrt(12) - df["sortino_6m"] = g["return_1m"].transform(lambda x: _rolling_sortino(x, window=6)) - df["calmar_proxy"] = _safe_div( - g["return_1m"].transform(lambda x: x.rolling(12, min_periods=6).mean() * 12), - g["return_1m"].transform(lambda x: _rolling_maxdrawdown(x, window=12)).abs(), - ) - return df - - -def _add_tail_risk_factors(df: pd.DataFrame, g) -> pd.DataFrame: - df["return_skew_6m"] = g["return_1m"].transform(lambda x: x.rolling(6, min_periods=3).skew()) - df["return_kurt_6m"] = g["return_1m"].transform(lambda x: x.rolling(6, min_periods=3).kurt()) - df["hist_var_5pct"] = g["return_1m"].transform(lambda x: x.rolling(12, min_periods=6).quantile(0.05)) - - def _cvar(r: np.ndarray) -> float: - t = np.quantile(r, 0.05) - tail = r[r <= t] - return tail.mean() if len(tail) > 0 else np.nan - - df["cvar_5pct"] = g["return_1m"].transform(lambda x: x.rolling(12, min_periods=6).apply(_cvar, raw=True)) - return df - - -def _add_technical_enrichment(df: pd.DataFrame, g) -> pd.DataFrame: - if "rsi" in df.columns: - df["rsi_divergence"] = ( - g["adj close"].transform(lambda x: x.pct_change(3)) - - g["rsi"].transform(lambda x: x.pct_change(3)) - ) - - if "euro_volume" in df.columns: - df["amihud_illiquidity"] = _safe_div(df["return_1m"].abs(), df["euro_volume"]) - df["volume_trend_3m"] = g["euro_volume"].transform(lambda x: x.pct_change(3)) - df["volume_zscore"] = g["euro_volume"].transform( - lambda x: _safe_div( - x - x.rolling(12, min_periods=6).mean(), - x.rolling(12, min_periods=6).std(), - ) - ) - else: - for col in ["amihud_illiquidity", "volume_trend_3m", "volume_zscore"]: - df[col] = 0.0 return df - -def _add_seasonality_features(df: pd.DataFrame) -> pd.DataFrame: - dates = df.index.get_level_values("date") - df["month_sin"] = np.sin(2 * np.pi * dates.month / 12) - df["month_cos"] = np.cos(2 * np.pi * dates.month / 12) - df["is_q_end"] = dates.month.isin([3, 6, 9, 12]).astype(int) - df["is_jan"] = (dates.month == 1).astype(int) - return df - - def get_fama_french_betas(data: pd.DataFrame) -> pd.DataFrame: logger.info("Retrieving Fama-French factors...") try: @@ -224,47 +106,44 @@ def get_fama_french_betas(data: pd.DataFrame) -> pd.DataFrame: for ticker in data.index.get_level_values(1).unique(): ticker_data = data.xs(ticker, level=1) y = ticker_data["return_1m"].dropna() - if y.empty: - continue + if y.empty: continue X = factor_data.loc[factor_data.index.intersection(y.index)] y = y.loc[X.index] - if len(y) <= MIN_HISTORY_FF: - continue + if len(y) <= MIN_HISTORY_FF: continue with warnings.catch_warnings(): warnings.filterwarnings("ignore", category=RuntimeWarning, message=".*divide by zero.*") - params = ( - RollingOLS(y, sm.add_constant(X[FAMA_FRENCH_FACTORS]), window=MIN_HISTORY_FF) - .fit() - .params - .drop("const", axis=1) - ) - + params = RollingOLS(y, sm.add_constant(X[FAMA_FRENCH_FACTORS]), window=MIN_HISTORY_FF).fit().params.drop("const", axis=1) params["ticker"] = ticker betas_list.append(params) if betas_list: betas_df = pd.concat(betas_list).set_index("ticker", append=True) data = data.join(betas_df.groupby("ticker").shift()) - data[FAMA_FRENCH_FACTORS] = ( - data.groupby(level="ticker", group_keys=False)[FAMA_FRENCH_FACTORS] - .transform(lambda x: x.fillna(x.mean())) - ) + data[FAMA_FRENCH_FACTORS] = data.groupby(level="ticker", group_keys=False)[FAMA_FRENCH_FACTORS].transform(lambda x: x.fillna(x.mean())) return data - except Exception as exc: logger.warning(f"Fama-French retrieval failed ({exc}).") - + return data.assign(**{f: 0.0 for f in FAMA_FRENCH_FACTORS}) +# ══════════════════════════════════════════════════════════════════ +# 2. ORCHESTRATION DES FEATURES +# ══════════════════════════════════════════════════════════════════ def add_all_features(df: pd.DataFrame) -> pd.DataFrame: if not isinstance(df.index, pd.MultiIndex): - raise ValueError("MultiIndex requis.") + raise ValueError("MultiIndex requis (date, ticker).") + df = df.copy() + + # 1. Calcul Fama-French avant tout (Solution au KeyError) + df = get_fama_french_betas(df) + g = df.groupby(level="ticker") - logger.info("Computing alpha features...") + + # 2. Autres calculs df = _add_momentum_factors(df, g) df = _add_mean_reversion_factors(df, g) df = _add_volatility_factors(df, g) @@ -274,15 +153,15 @@ def add_all_features(df: pd.DataFrame) -> pd.DataFrame: df = _add_seasonality_features(df) df = add_rank_features(df) - cols_to_lag = [ - "rsi", "macd", "bb_low", "bb_mid", "bb_high", "atr", - "garman_klass_vol", "bb_position", "macd_sign", - "Mkt-RF", "SMB", "HML", "RMW", "CMA", - ] + # 3. Lags + cols_to_lag = ["rsi", "macd", "bb_low", "bb_mid", "bb_high", "atr", "garman_klass_vol", "bb_position", "macd_sign"] + FAMA_FRENCH_FACTORS for col in cols_to_lag: - if col in df.columns: - df[f"{col}_lag1"] = df.groupby(level="ticker")[col].shift(1) + if col not in df.columns: + df[col] = 0.0 + df[f"{col}_lag1"] = df.groupby(level="ticker")[col].shift(1) df = df.fillna(0).replace([np.inf, -np.inf], 0) - logger.info(f" Features prêtes. Shape : {df.shape}") - return df \ No newline at end of file + logger.info(f"Features prêtes. Shape : {df.shape}") + return df + +# Fonctions internes omises pour la concision (utilise les mêmes que ton ancien fichier) From e24af0f287d4fcc30fbb892872af6354e845a110 Mon Sep 17 00:00:00 2001 From: MoussaTheAnalyst <163172464+SORADATA@users.noreply.github.com> Date: Tue, 7 Jul 2026 16:01:37 +0200 Subject: [PATCH 13/17] feat: Add calculate_returns function for monthly returns Added a new function to calculate monthly returns based on 'adj close'. This function ensures compatibility with MarketDataProcessor. --- src/features/alpha_features.py | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/src/features/alpha_features.py b/src/features/alpha_features.py index 486257a..8b63c63 100644 --- a/src/features/alpha_features.py +++ b/src/features/alpha_features.py @@ -163,5 +163,14 @@ def add_all_features(df: pd.DataFrame) -> pd.DataFrame: df = df.fillna(0).replace([np.inf, -np.inf], 0) logger.info(f"Features prêtes. Shape : {df.shape}") return df - -# Fonctions internes omises pour la concision (utilise les mêmes que ton ancien fichier) +# Ajoute ceci à la fin de src/features/alpha_features.py + +def calculate_returns(df: pd.DataFrame) -> pd.DataFrame: + """ + Fonction wrapper pour assurer la compatibilité avec MarketDataProcessor. + Calcule les rendements mensuels basés sur 'adj close'. + """ + # On s'assure de travailler sur une copie pour éviter les problèmes d'index + df = df.copy() + g = df.groupby(level="ticker") + return _add_momentum_factors(df, g) From bc64163ec6c6bf9f94db274885e40bebddd39b30 Mon Sep 17 00:00:00 2001 From: MoussaTheAnalyst <163172464+SORADATA@users.noreply.github.com> Date: Tue, 7 Jul 2026 16:09:02 +0200 Subject: [PATCH 14/17] Refactor: Remove calculate_returns function from alpha_features.py Remove calculate_returns function and its comments. --- src/features/alpha_features.py | 12 +----------- 1 file changed, 1 insertion(+), 11 deletions(-) diff --git a/src/features/alpha_features.py b/src/features/alpha_features.py index 8b63c63..8e954f7 100644 --- a/src/features/alpha_features.py +++ b/src/features/alpha_features.py @@ -163,14 +163,4 @@ def add_all_features(df: pd.DataFrame) -> pd.DataFrame: df = df.fillna(0).replace([np.inf, -np.inf], 0) logger.info(f"Features prêtes. Shape : {df.shape}") return df -# Ajoute ceci à la fin de src/features/alpha_features.py - -def calculate_returns(df: pd.DataFrame) -> pd.DataFrame: - """ - Fonction wrapper pour assurer la compatibilité avec MarketDataProcessor. - Calcule les rendements mensuels basés sur 'adj close'. - """ - # On s'assure de travailler sur une copie pour éviter les problèmes d'index - df = df.copy() - g = df.groupby(level="ticker") - return _add_momentum_factors(df, g) + From 6c6b903c0f9ae2d2071d6766f39dedbad0c4fad7 Mon Sep 17 00:00:00 2001 From: MoussaTheAnalyst <163172464+SORADATA@users.noreply.github.com> Date: Tue, 7 Jul 2026 16:12:13 +0200 Subject: [PATCH 15/17] Refactor: cleaning comments and updating logic Removed commented-out documentation and unnecessary comments. Updated the processing pipeline to include feature calculations and cleaned up the code for better readability. --- src/transform/processor.py | 60 ++++---------------------------------- 1 file changed, 5 insertions(+), 55 deletions(-) diff --git a/src/transform/processor.py b/src/transform/processor.py index cdbbea8..dd9d756 100644 --- a/src/transform/processor.py +++ b/src/transform/processor.py @@ -1,27 +1,12 @@ -""" -MarketDataProcessor -==================== -Pipeline de transformation complète : - 1. Validation & nettoyage des tickers - 2. Indicateurs techniques (daily) - 3. Agrégation mensuelle - 4. Returns winsorisés (mensuel) - 5. Betas Fama-French rolling (mensuel) - 6. Alpha features - 7. Lag des variables (anti-leakage) -""" - import pandas as pd from typing import Tuple, List from const import VARS_TO_LAG, RESAMPLE_MEAN_COLS, RESAMPLE_LAST_EXCLUDE -# processor.py — imports corrects from src.features.alpha_features import ( compute_technical_indicators, - calculate_returns, get_fama_french_betas, + add_all_features, ) -from src.features.alpha_features import add_all_features from src.transform.ticker_manager import validate_and_clean_tickers from src.utils.logger import setup_logger @@ -46,9 +31,6 @@ def __init__(self, active_tickers: List[str]): def _resample_to_monthly(self, df: pd.DataFrame) -> pd.DataFrame: """ Agrège les données journalières en mensuel (Business Month End). - - - euro_volume : moyenne mensuelle (RESAMPLE_MEAN_COLS) - - autres cols : dernière valeur du mois (last) """ last_cols = [c for c in df.columns if c not in RESAMPLE_LAST_EXCLUDE] @@ -65,21 +47,10 @@ def _resample_to_monthly(self, df: pd.DataFrame) -> pd.DataFrame: ) monthly = pd.concat([mean_part, last_part], axis=1).dropna(how="all") - - # S'assurer que l'index est bien ordonné (date, ticker) - monthly = monthly.sort_index() - - return monthly + return monthly.sort_index() # Lag des variables macro/volume def _apply_lags(self, df: pd.DataFrame) -> pd.DataFrame: - """ - Lag d'une période pour toutes les variables dans VARS_TO_LAG. - - Note : les lags des indicateurs TA (rsi, macd, bb_*, atr, cluster) - sont gérés directement dans alpha_features._lag_ta_indicators() - pour éviter la duplication. - """ for col in VARS_TO_LAG: if col in df.columns: df[f"{col}_lag1"] = df.groupby(level="ticker")[col].shift(1) @@ -89,47 +60,26 @@ def _apply_lags(self, df: pd.DataFrame) -> pd.DataFrame: def process(self, raw_df: pd.DataFrame) -> Tuple[pd.DataFrame, pd.DataFrame, dict]: """ Exécute la pipeline de transformation complète. - - Parameters - ---------- - raw_df : pd.DataFrame - Données brutes multi-index (date, ticker) issues de l'extracteur. - - Returns - ------- - df_daily : pd.DataFrame — données journalières enrichies - df_monthly : pd.DataFrame — données mensuelles avec toutes les features - alerts : dict — alertes de validation des tickers """ logger.info("Début du processing des données...") df = raw_df.copy() if "adj close" not in df.columns and "close" in df.columns: df["adj close"] = df["close"] logger.warning("adj close absent — utilisation de close comme proxy.") + df, valid_tickers, alerts = validate_and_clean_tickers(df, self.active_tickers) - df = compute_technical_indicators(df) logger.info("Agrégation à la fréquence mensuelle...") df_monthly = self._resample_to_monthly(df) - df_monthly = df_monthly.groupby( - level=1, group_keys=False - ).apply(calculate_returns) - + # Calcul des features et facteurs df_monthly = get_fama_french_betas(df_monthly) - - # alpha features df_monthly = add_all_features(df_monthly) - df_monthly = self._apply_lags(df_monthly) - # Final verif n_features = df_monthly.shape[1] n_obs = len(df_monthly) - logger.info( - f"Processing terminé avec succès. " - f"Monthly shape : ({n_obs}, {n_features})" - ) + logger.info(f"Processing terminé. Monthly shape : ({n_obs}, {n_features})") return df, df_monthly, alerts From 7b846617f600058952739f77873a4f8c3148cd0f Mon Sep 17 00:00:00 2001 From: MoussaTheAnalyst <163172464+SORADATA@users.noreply.github.com> Date: Tue, 7 Jul 2026 16:16:28 +0200 Subject: [PATCH 16/17] feat: Add newline at end of processor.py Ensure newline at end of file for consistency. From fde52cc56a0feac9f9c098232471aeaa83debad7 Mon Sep 17 00:00:00 2001 From: MoussaTheAnalyst <163172464+SORADATA@users.noreply.github.com> Date: Tue, 7 Jul 2026 16:24:00 +0200 Subject: [PATCH 17/17] Refactor: clarity and efficiency Refactor and enhance alpha feature calculations with improved readability and structure. --- src/features/alpha_features.py | 144 ++++++++++++++++++--------------- 1 file changed, 81 insertions(+), 63 deletions(-) diff --git a/src/features/alpha_features.py b/src/features/alpha_features.py index 8e954f7..dc5e2ae 100644 --- a/src/features/alpha_features.py +++ b/src/features/alpha_features.py @@ -18,7 +18,7 @@ logger = setup_logger("alpha_features") # ══════════════════════════════════════════════════════════════════ -# UTILITAIRES +# 1. SOUS-FONCTIONS MATHÉMATIQUES (Utils) # ══════════════════════════════════════════════════════════════════ def _safe_div(a: pd.Series, b: pd.Series) -> pd.Series: @@ -27,8 +27,7 @@ def _safe_div(a: pd.Series, b: pd.Series) -> pd.Series: def _rolling_sortino(returns: pd.Series, window: int = 6) -> pd.Series: def _sortino_scalar(r: np.ndarray) -> float: neg = r[r < 0] - if len(neg) == 0 or np.std(neg) == 0: - return np.nan + if len(neg) == 0 or np.std(neg) == 0: return np.nan return (np.mean(r) / np.std(neg)) * np.sqrt(12) return returns.rolling(window, min_periods=window // 2).apply(_sortino_scalar, raw=True) @@ -39,53 +38,100 @@ def _mdd(r: np.ndarray) -> float: return ((cumulative - peak) / peak).min() return returns.rolling(window, min_periods=window // 2).apply(_mdd, raw=True) +# ══════════════════════════════════════════════════════════════════ +# 2. CALCULS DES FEATURES (Logique métier) +# ══════════════════════════════════════════════════════════════════ + +def _add_momentum_factors(df: pd.DataFrame, g) -> pd.DataFrame: + df["return_1m"] = g["adj close"].transform(lambda x: x.pct_change(1)) + for lag in [2, 3, 6, 9, 12]: + df[f"return_{lag}m"] = g["adj close"].transform(lambda x: x.pct_change(lag)) + pct_12 = g["adj close"].transform(lambda x: x.pct_change(12)) + pct_1 = g["adj close"].transform(lambda x: x.pct_change(1)) + pct_6 = g["adj close"].transform(lambda x: x.pct_change(6)) + df["mom_12_1"] = pct_12.div(pct_1.replace(0, np.nan)).replace([np.inf, -np.inf], np.nan) + df["mom_6_1"] = pct_6.div(pct_1.replace(0, np.nan)).replace([np.inf, -np.inf], np.nan) + df["mom_3_1"] = g["adj close"].transform(lambda x: x.pct_change(3)) + return df + +def _add_mean_reversion_factors(df: pd.DataFrame, g) -> pd.DataFrame: + ma12 = g["adj close"].transform(lambda x: x.rolling(12, min_periods=6).mean()) + std12 = g["adj close"].transform(lambda x: x.rolling(12, min_periods=6).std()) + df["price_zscore_12"] = _safe_div(df["adj close"] - ma12, std12) + high52 = g["adj close"].transform(lambda x: x.rolling(12, min_periods=6).max()) + df["nearness_52w_high"] = _safe_div(df["adj close"], high52) + return df + +def _add_volatility_factors(df: pd.DataFrame, g) -> pd.DataFrame: + df["realized_vol_3m"] = g["return_1m"].transform(lambda x: x.rolling(3, min_periods=2).std() * np.sqrt(12)) + df["realized_vol_12m"] = g["return_1m"].transform(lambda x: x.rolling(12, min_periods=6).std() * np.sqrt(12)) + df["vol_ratio"] = df["realized_vol_3m"].div(df["realized_vol_12m"]).replace([np.inf, -np.inf], np.nan) + excess = df["return_1m"] - df["Mkt-RF"].fillna(0) if "Mkt-RF" in df.columns else df["return_1m"] + df["idio_vol"] = excess.groupby(level=1).transform(lambda x: x.rolling(6, min_periods=3).std() * np.sqrt(12)) + return df + +def _add_risk_adjusted_factors(df: pd.DataFrame, g) -> pd.DataFrame: + df["sharpe_3m"] = _safe_div(g["return_1m"].transform(lambda x: x.rolling(3, min_periods=2).mean()), g["return_1m"].transform(lambda x: x.rolling(3, min_periods=2).std())) * np.sqrt(12) + df["sharpe_6m"] = _safe_div(g["return_1m"].transform(lambda x: x.rolling(6, min_periods=3).mean()), g["return_1m"].transform(lambda x: x.rolling(6, min_periods=3).std())) * np.sqrt(12) + df["sortino_6m"] = g["return_1m"].transform(lambda x: _rolling_sortino(x, window=6)) + df["calmar_proxy"] = _safe_div(g["return_1m"].transform(lambda x: x.rolling(12, min_periods=6).mean() * 12), g["return_1m"].transform(lambda x: _rolling_maxdrawdown(x, window=12)).abs()) + return df + +def _add_tail_risk_factors(df: pd.DataFrame, g) -> pd.DataFrame: + df["return_skew_6m"] = g["return_1m"].transform(lambda x: x.rolling(6, min_periods=3).skew()) + df["return_kurt_6m"] = g["return_1m"].transform(lambda x: x.rolling(6, min_periods=3).kurt()) + df["hist_var_5pct"] = g["return_1m"].transform(lambda x: x.rolling(12, min_periods=6).quantile(0.05)) + def _cvar(r: np.ndarray) -> float: + t = np.quantile(r, 0.05); tail = r[r <= t] + return tail.mean() if len(tail) > 0 else np.nan + df["cvar_5pct"] = g["return_1m"].transform(lambda x: x.rolling(12, min_periods=6).apply(_cvar, raw=True)) + return df + +def _add_technical_enrichment(df: pd.DataFrame, g) -> pd.DataFrame: + if "rsi" in df.columns: + df["rsi_divergence"] = g["adj close"].transform(lambda x: x.pct_change(3)) - g["rsi"].transform(lambda x: x.pct_change(3)) + if "euro_volume" in df.columns: + df["amihud_illiquidity"] = _safe_div(df["return_1m"].abs(), df["euro_volume"]) + df["volume_trend_3m"] = g["euro_volume"].transform(lambda x: x.pct_change(3)) + df["volume_zscore"] = g["euro_volume"].transform(lambda x: _safe_div(x - x.rolling(12, min_periods=6).mean(), x.rolling(12, min_periods=6).std())) + return df + +def _add_seasonality_features(df: pd.DataFrame) -> pd.DataFrame: + dates = df.index.get_level_values("date") + df["month_sin"] = np.sin(2 * np.pi * dates.month / 12) + df["month_cos"] = np.cos(2 * np.pi * dates.month / 12) + df["is_q_end"] = dates.month.isin([3, 6, 9, 12]).astype(int) + return df + def add_rank_features(df: pd.DataFrame) -> pd.DataFrame: - features_to_rank = [ - "mom_12_1", "mom_6_1", "sharpe_6m", "sortino_6m", - "realized_vol_3m", "realized_vol_12m", - "amihud_illiquidity", "return_skew_6m", "hist_var_5pct", - ] + features_to_rank = ["mom_12_1", "mom_6_1", "sharpe_6m", "sortino_6m", "realized_vol_3m", "realized_vol_12m", "amihud_illiquidity", "return_skew_6m", "hist_var_5pct"] for feat in features_to_rank: - rank_col = f"{feat}_rank" - if feat in df.columns: - df[rank_col] = df.groupby(level="date")[feat].transform(lambda x: x.rank(pct=True)) - else: - df[rank_col] = 0.5 - return df.fillna({f"{f}_rank": 0.5 for f in features_to_rank}) + if feat in df.columns: df[f"{feat}_rank"] = df.groupby(level="date")[feat].transform(lambda x: x.rank(pct=True)) + else: df[f"{feat}_rank"] = 0.5 + return df # ══════════════════════════════════════════════════════════════════ -# 1. CALCULS TECHNIQUES & FACTEURS +# 3. FONCTIONS PRINCIPALES (Exposées) # ══════════════════════════════════════════════════════════════════ def compute_technical_indicators(df: pd.DataFrame) -> pd.DataFrame: logger.info("Computing daily technical indicators...") if all(col in df.columns for col in ["high", "low", "open", "adj close"]): - df["garman_klass_vol"] = ( - (np.log(df["high"]) - np.log(df["low"])) ** 2 / 2 - - (2 * np.log(2) - 1) * (np.log(df["adj close"]) - np.log(df["open"])) ** 2 - ) + df["garman_klass_vol"] = ((np.log(df["high"]) - np.log(df["low"])) ** 2 / 2 - (2 * np.log(2) - 1) * (np.log(df["adj close"]) - np.log(df["open"])) ** 2) else: df["garman_klass_vol"] = 0.0 - for ticker in df.index.get_level_values(1).unique(): idx = (slice(None), ticker) close = df.loc[idx, "adj close"] if len(close) > MIN_HISTORY_TA: df.loc[idx, "rsi"] = RSIIndicator(close=close, window=RSI_WINDOW).rsi().values bb = BollingerBands(close=np.log1p(close), window=BB_WINDOW, window_dev=BB_STD) - df.loc[idx, "bb_low"] = bb.bollinger_lband().values - df.loc[idx, "bb_mid"] = bb.bollinger_mavg().values - df.loc[idx, "bb_high"] = bb.bollinger_hband().values + df.loc[idx, "bb_low"], df.loc[idx, "bb_mid"], df.loc[idx, "bb_high"] = bb.bollinger_lband().values, bb.bollinger_mavg().values, bb.bollinger_hband().values df.loc[idx, "bb_position"] = ((np.log1p(close) - bb.bollinger_lband()) / (bb.bollinger_hband() - bb.bollinger_lband() + 1e-9)) - df["atr"] = df.groupby(level=1, group_keys=False).apply(compute_atr) df["macd"] = df.groupby(level=1, group_keys=False).apply(compute_macd) df["macd_sign"] = np.sign(df["macd"].fillna(0)) - - if "volume" in df.columns: - df["euro_volume"] = (df["adj close"] * df["volume"]) / 1e6 - else: - df["euro_volume"] = 0.0 + df["euro_volume"] = (df["adj close"] * df["volume"]) / 1e6 if "volume" in df.columns else 0.0 return df def get_fama_french_betas(data: pd.DataFrame) -> pd.DataFrame: @@ -94,14 +140,10 @@ def get_fama_french_betas(data: pd.DataFrame) -> pd.DataFrame: with warnings.catch_warnings(): warnings.filterwarnings("ignore", message=".*date_parser.*") factor_data = web.DataReader("Europe_5_Factors", "famafrench", start="2010")[0].drop("RF", axis=1) - factor_data.index = pd.to_datetime(factor_data.index.to_timestamp()).tz_localize(None) factor_data = factor_data.resample("BME").last().div(100) factor_data.index.name = "date" - - if "return_1m" not in data.columns: - return data - + if "return_1m" not in data.columns: return data betas_list = [] for ticker in data.index.get_level_values(1).unique(): ticker_data = data.xs(ticker, level=1) @@ -110,13 +152,9 @@ def get_fama_french_betas(data: pd.DataFrame) -> pd.DataFrame: X = factor_data.loc[factor_data.index.intersection(y.index)] y = y.loc[X.index] if len(y) <= MIN_HISTORY_FF: continue - - with warnings.catch_warnings(): - warnings.filterwarnings("ignore", category=RuntimeWarning, message=".*divide by zero.*") - params = RollingOLS(y, sm.add_constant(X[FAMA_FRENCH_FACTORS]), window=MIN_HISTORY_FF).fit().params.drop("const", axis=1) + params = RollingOLS(y, sm.add_constant(X[FAMA_FRENCH_FACTORS]), window=MIN_HISTORY_FF).fit().params.drop("const", axis=1) params["ticker"] = ticker betas_list.append(params) - if betas_list: betas_df = pd.concat(betas_list).set_index("ticker", append=True) data = data.join(betas_df.groupby("ticker").shift()) @@ -124,26 +162,13 @@ def get_fama_french_betas(data: pd.DataFrame) -> pd.DataFrame: return data except Exception as exc: logger.warning(f"Fama-French retrieval failed ({exc}).") - return data.assign(**{f: 0.0 for f in FAMA_FRENCH_FACTORS}) -# ══════════════════════════════════════════════════════════════════ -# 2. ORCHESTRATION DES FEATURES -# ══════════════════════════════════════════════════════════════════ - def add_all_features(df: pd.DataFrame) -> pd.DataFrame: - if not isinstance(df.index, pd.MultiIndex): - raise ValueError("MultiIndex requis (date, ticker).") - - df = df.copy() - - # 1. Calcul Fama-French avant tout (Solution au KeyError) - df = get_fama_french_betas(df) - + if not isinstance(df.index, pd.MultiIndex): raise ValueError("MultiIndex requis.") + df = get_fama_french_betas(df.copy()) g = df.groupby(level="ticker") logger.info("Computing alpha features...") - - # 2. Autres calculs df = _add_momentum_factors(df, g) df = _add_mean_reversion_factors(df, g) df = _add_volatility_factors(df, g) @@ -152,15 +177,8 @@ def add_all_features(df: pd.DataFrame) -> pd.DataFrame: df = _add_technical_enrichment(df, g) df = _add_seasonality_features(df) df = add_rank_features(df) - - # 3. Lags cols_to_lag = ["rsi", "macd", "bb_low", "bb_mid", "bb_high", "atr", "garman_klass_vol", "bb_position", "macd_sign"] + FAMA_FRENCH_FACTORS for col in cols_to_lag: - if col not in df.columns: - df[col] = 0.0 + if col not in df.columns: df[col] = 0.0 df[f"{col}_lag1"] = df.groupby(level="ticker")[col].shift(1) - - df = df.fillna(0).replace([np.inf, -np.inf], 0) - logger.info(f"Features prêtes. Shape : {df.shape}") - return df - + return df.fillna(0).replace([np.inf, -np.inf], 0)