From 2001756bad765258fb112c3187ba2da3958fb08a Mon Sep 17 00:00:00 2001 From: sora Date: Tue, 7 Jul 2026 15:10:16 +0000 Subject: [PATCH 01/13] fix: Make scoring best for unused features --- src/pipeline/backtest.py | 97 ++++++++++++---------------------------- 1 file changed, 28 insertions(+), 69 deletions(-) diff --git a/src/pipeline/backtest.py b/src/pipeline/backtest.py index 8596de1..2da9273 100644 --- a/src/pipeline/backtest.py +++ b/src/pipeline/backtest.py @@ -33,108 +33,76 @@ # 1. CONSTRUCTION DE PORTEFEUILLE & HELPERS # ============================================================================= -def get_optimal_weights( - prices_df: pd.DataFrame, - risk_free_rate: float = RISK_FREE_RATE, -) -> Tuple[Dict[str, float], str]: +def get_optimal_weights(prices_df: pd.DataFrame, risk_free_rate: float = RISK_FREE_RATE) -> Tuple[Dict[str, float], str]: 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) cov = risk_models.CovarianceShrinkage(prices_df, frequency=TRADING_DAYS_YEAR).ledoit_wolf() - 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) - - weights = dict(ef.clean_weights()) - _log_concentration(weights) - return weights, "max_sharpe" + return dict(ef.clean_weights()), "max_sharpe" except Exception as exc: 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: - 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: + """Scoring robuste avec injection automatique des features manquantes.""" + expected_cols = _resolve_expected_features(model) + # On force l'alignement : si features manquantes, elles sont créées à 0 ici + x_input = features.reindex(columns=expected_cols).fillna(0) + if hasattr(model, "predict_proba"): - expected_cols = getattr(model, "features_", None) - 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() + return np.asarray(model.predict(x_input)).ravel() + +def _resolve_expected_features(model: Any) -> Optional[List[str]]: + """Résout la liste de features réellement attendue par le modèle (v5 ou v6).""" + try: + inner = model._model_impl.python_model + if hasattr(inner, "features_"): return list(inner.features_) + except Exception: pass + try: + input_schema = model.metadata.get_input_schema() + if input_schema: return [c.name for c in input_schema.inputs] + except Exception: pass + return None def _build_price_matrix(df_daily: pd.DataFrame, ffill_limit: int = MAX_PRICE_FFILL_DAYS) -> pd.DataFrame: - if "adj_close" in df_daily.columns: - col = "adj_close" - elif "adj close" in df_daily.columns: - col = "adj close" - else: - raise KeyError("Colonnes de prix non trouvées.") + col = "adj_close" if "adj_close" in df_daily.columns else "adj close" 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 + return df_feat.xs(last_date, level="date").copy(), last_date # ============================================================================= -# 2. MOTEUR DE SIMULATION & ANALYTICS +# 2. MOTEUR DE SIMULATION & API # ============================================================================= -def _simulate_period( - allocation: Dict[str, float], - drifted_allocation: Dict[str, float], - trading_days: pd.DatetimeIndex, - daily_returns: pd.DataFrame, - benchmark_returns: pd.Series, - portfolio_value: float, - benchmark_value: float, -) -> Tuple[pd.DataFrame, float, float, Dict[str, float]]: - +def _simulate_period(allocation, drifted_allocation, trading_days, daily_returns, benchmark_returns, portfolio_value, benchmark_value): 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()) 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 -# ============================================================================= -# 3. API PUBLIQUE -# ============================================================================= - -def backtest_strategy_with_rebalancing( - df_daily: pd.DataFrame, df_monthly: pd.DataFrame, model: Any, benchmark_ticker: str -) -> Tuple[pd.DataFrame, pd.DataFrame, Dict[str, float]]: - +def backtest_strategy_with_rebalancing(df_daily, df_monthly, model, benchmark_ticker): df_monthly_feat = add_all_features(df_monthly.copy()) daily_prices = _build_price_matrix(df_daily) daily_returns = daily_prices.pct_change().fillna(0) - 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) portfolio_value, drifted_allocation, period_frames, rebalance_log = 100.0, {}, [], [] @@ -143,40 +111,31 @@ def backtest_strategy_with_rebalancing( for i, month_date in enumerate(monthly_dates[:-1]): 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, "Allocation": allocation}) + return pd.concat(period_frames), pd.DataFrame(rebalance_log).set_index("Date"), {} - 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) -> Tuple[pd.DataFrame, pd.DataFrame]: +def generate_live_signals(df_daily, daily_prices, model, rebalance_history): snapshot, last_date = _build_daily_snapshot(df_daily) 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: 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, _ = get_optimal_weights(prices_subset) allocation = {t: w for t, w in weights.items() if w > 1e-4} - 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 + return out[["Ticker", "Signal", "Allocation", "Proba_Hausse"]], rebalance_history \ No newline at end of file From c62ae1e624c095399cbdfe86dc9ee1a207f1679a Mon Sep 17 00:00:00 2001 From: sora Date: Tue, 7 Jul 2026 15:11:30 +0000 Subject: [PATCH 02/13] fix: Add is_jan --- src/features/alpha_features.py | 1 + 1 file changed, 1 insertion(+) diff --git a/src/features/alpha_features.py b/src/features/alpha_features.py index e8f4380..0e95e48 100644 --- a/src/features/alpha_features.py +++ b/src/features/alpha_features.py @@ -101,6 +101,7 @@ def _add_seasonality_features(df: pd.DataFrame) -> pd.DataFrame: 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 add_rank_features(df: pd.DataFrame) -> pd.DataFrame: From d96d67626113885e74884543b94bc93b8b9a62be Mon Sep 17 00:00:00 2001 From: sora Date: Tue, 7 Jul 2026 16:33:17 +0000 Subject: [PATCH 03/13] refactor: Add logger to _x_prepared --- src/models/ensemble.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/src/models/ensemble.py b/src/models/ensemble.py index adf270f..1d21b88 100644 --- a/src/models/ensemble.py +++ b/src/models/ensemble.py @@ -46,16 +46,18 @@ def _available(df: pd.DataFrame) -> List[str]: def _prepare_X(df: pd.DataFrame, features: List[str]) -> pd.DataFrame: df_prepared = df.copy() - for col in features: - if col not in df_prepared.columns: + missing = [col for col in features if col not in df_prepared.columns] + if missing: + logger.warning(f"Features manquantes, remplies à 0.0 : {missing}") + for col in missing: df_prepared[col] = 0.0 - return df[features].fillna(0).replace([np.inf, -np.inf], 0) - + return df_prepared[features].fillna(0).replace([np.inf, -np.inf], 0) # ══════════════════════════════════════════════════════════════════ # OPTUNA OBJECTIVES # ══════════════════════════════════════════════════════════════════ + def _xgb_objective(X: pd.DataFrame, y: pd.Series, n_trials: int) -> xgb.XGBClassifier: from sklearn.metrics import roc_auc_score cv = PurgedTimeSeriesSplit(n_splits=5) From 26531ae732f9d5d3924a6c449a98edfcf93621fa Mon Sep 17 00:00:00 2001 From: sora Date: Tue, 7 Jul 2026 17:28:15 +0000 Subject: [PATCH 04/13] feat: Add maths import moduls --- src/features/alpha_features.py | 38 ++++++++++++++-------------------- 1 file changed, 16 insertions(+), 22 deletions(-) diff --git a/src/features/alpha_features.py b/src/features/alpha_features.py index 0e95e48..2297e1a 100644 --- a/src/features/alpha_features.py +++ b/src/features/alpha_features.py @@ -13,35 +13,18 @@ FAMA_FRENCH_FACTORS, ) from src.utils.feature_utils import compute_atr, compute_macd +from src.utils.math_utils import _safe_div, _rolling_sortino, _rolling_maxdrawdown from src.utils.logger import setup_logger -logger = setup_logger("alpha_features") - -# ══════════════════════════════════════════════════════════════════ -# 1. SOUS-FONCTIONS MATHÉMATIQUES (Utils) -# ══════════════════════════════════════════════════════════════════ - -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] - 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) +logger = setup_logger("alpha_features") -def _rolling_maxdrawdown(returns: pd.Series, window: int = 12) -> pd.Series: - def _mdd(r: np.ndarray) -> float: - cumulative = np.cumprod(1 + r) - peak = np.maximum.accumulate(cumulative) - return ((cumulative - peak) / peak).min() - return returns.rolling(window, min_periods=window // 2).apply(_mdd, raw=True) # ══════════════════════════════════════════════════════════════════ -# 2. CALCULS DES FEATURES (Logique métier) +# 1. 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]: @@ -54,6 +37,7 @@ def _add_momentum_factors(df: pd.DataFrame, g) -> pd.DataFrame: 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()) @@ -62,6 +46,7 @@ def _add_mean_reversion_factors(df: pd.DataFrame, g) -> pd.DataFrame: 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)) @@ -70,6 +55,7 @@ def _add_volatility_factors(df: pd.DataFrame, g) -> pd.DataFrame: 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) @@ -77,6 +63,7 @@ def _add_risk_adjusted_factors(df: pd.DataFrame, g) -> pd.DataFrame: 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()) @@ -87,6 +74,7 @@ def _cvar(r: np.ndarray) -> float: 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)) @@ -96,6 +84,7 @@ def _add_technical_enrichment(df: pd.DataFrame, g) -> pd.DataFrame: 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) @@ -104,6 +93,7 @@ def _add_seasonality_features(df: pd.DataFrame) -> pd.DataFrame: df["is_jan"] = (dates.month == 1).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"] for feat in features_to_rank: @@ -111,10 +101,12 @@ def add_rank_features(df: pd.DataFrame) -> pd.DataFrame: else: df[f"{feat}_rank"] = 0.5 return df + # ══════════════════════════════════════════════════════════════════ -# 3. FONCTIONS PRINCIPALES (Exposées) +# 2. 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"]): @@ -135,6 +127,7 @@ def compute_technical_indicators(df: pd.DataFrame) -> pd.DataFrame: 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: logger.info("Retrieving Fama-French factors...") try: @@ -165,6 +158,7 @@ def get_fama_french_betas(data: pd.DataFrame) -> pd.DataFrame: logger.warning(f"Fama-French retrieval failed ({exc}).") return data.assign(**{f: 0.0 for f in FAMA_FRENCH_FACTORS}) + def add_all_features(df: pd.DataFrame) -> pd.DataFrame: if not isinstance(df.index, pd.MultiIndex): raise ValueError("MultiIndex requis.") df = get_fama_french_betas(df.copy()) From 40ff98aa280cfd8e509e5fd0e9cab88a1cd1a480 Mon Sep 17 00:00:00 2001 From: sora Date: Tue, 7 Jul 2026 17:28:49 +0000 Subject: [PATCH 05/13] feat: Add math utils --- src/utils/math_utils.py | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) create mode 100644 src/utils/math_utils.py diff --git a/src/utils/math_utils.py b/src/utils/math_utils.py new file mode 100644 index 0000000..724e06a --- /dev/null +++ b/src/utils/math_utils.py @@ -0,0 +1,22 @@ +import numpy as np +import pandas as pd + + +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] + 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) + + +def _rolling_maxdrawdown(returns: pd.Series, window: int = 12) -> pd.Series: + def _mdd(r: np.ndarray) -> float: + cumulative = np.cumprod(1 + r) + peak = np.maximum.accumulate(cumulative) + return ((cumulative - peak) / peak).min() + return returns.rolling(window, min_periods=window // 2).apply(_mdd, raw=True) \ No newline at end of file From 94b50d55adedc5169db945ca55203393ed471d70 Mon Sep 17 00:00:00 2001 From: sora Date: Tue, 7 Jul 2026 17:30:04 +0000 Subject: [PATCH 06/13] refactor: benckmark_value --- src/pipeline/backtest.py | 64 ++++++++++++++++++++++++++++++++++++---- 1 file changed, 58 insertions(+), 6 deletions(-) diff --git a/src/pipeline/backtest.py b/src/pipeline/backtest.py index 2da9273..4a0b9ae 100644 --- a/src/pipeline/backtest.py +++ b/src/pipeline/backtest.py @@ -29,6 +29,15 @@ MAX_PRICE_FFILL_DAYS = 5 MAX_ACCEPTABLE_MISSING_RATIO = 0.05 +# Nombre minimum de mois d'historique requis par ticker avant que les features +# a fenetre glissante (jusqu'a 12 mois pour realized_vol_12m, sharpe_6m, +# calmar_proxy, hist_var_5pct, cvar_5pct...) soient jugees fiables. Les mois +# en-dessous de ce seuil sont scores sur des features encore incompletes, +# masquees a zero par le fillna(0) interne d'add_all_features -> cela produit +# une periode plate artificielle en debut de backtest si on ne filtre pas. +MIN_FEATURE_HISTORY = 12 + + # ============================================================================= # 1. CONSTRUCTION DE PORTEFEUILLE & HELPERS # ============================================================================= @@ -48,16 +57,18 @@ def get_optimal_weights(prices_df: pd.DataFrame, risk_free_rate: float = RISK_FR logger.warning(f"Max Sharpe a échoué ({exc}) -> fallback.") return {t: 1.0 / n_assets for t in prices_df.columns}, "equal_weight" + def _score_with_model(model: Any, features: pd.DataFrame) -> np.ndarray: """Scoring robuste avec injection automatique des features manquantes.""" expected_cols = _resolve_expected_features(model) # On force l'alignement : si features manquantes, elles sont créées à 0 ici x_input = features.reindex(columns=expected_cols).fillna(0) - + if hasattr(model, "predict_proba"): return model.predict_proba(x_input)[:, 1] return np.asarray(model.predict(x_input)).ravel() + def _resolve_expected_features(model: Any) -> Optional[List[str]]: """Résout la liste de features réellement attendue par le modèle (v5 ou v6).""" try: @@ -70,16 +81,32 @@ def _resolve_expected_features(model: Any) -> Optional[List[str]]: except Exception: pass return None + def _build_price_matrix(df_daily: pd.DataFrame, ffill_limit: int = MAX_PRICE_FFILL_DAYS) -> pd.DataFrame: col = "adj_close" if "adj_close" in df_daily.columns else "adj close" return df_daily[col].unstack().ffill(limit=ffill_limit) + def _build_daily_snapshot(df_daily: pd.DataFrame) -> Tuple[pd.DataFrame, pd.Timestamp]: last_date = df_daily.index.get_level_values("date").max() - lookback_df = df_daily.iloc[-252:].copy() + lookback_df = df_daily.iloc[-252:].copy() df_feat = add_all_features(lookback_df) return df_feat.xs(last_date, level="date").copy(), last_date + +def _filter_warmup_period(df_monthly: pd.DataFrame, min_history: int = MIN_FEATURE_HISTORY) -> pd.Series: + """ + Calcule un masque booleen (indexe comme df_monthly) qui exclut, pour + chaque ticker, les tout premiers mois n'ayant pas assez d'historique pour + que les features a fenetre glissante soient completes. Le calcul se base + sur le nombre de lignes deja observees par ticker (donnees brutes, avant + tout calcul de features), donc independant du fillna(0) applique plus + tard par add_all_features. + """ + history_count = df_monthly.groupby(level="ticker").cumcount() + 1 + return history_count > min_history + + # ============================================================================= # 2. MOTEUR DE SIMULATION & API # ============================================================================= @@ -99,15 +126,33 @@ def _simulate_period(allocation, drifted_allocation, trading_days, daily_returns 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 + def backtest_strategy_with_rebalancing(df_daily, df_monthly, model, benchmark_ticker): + # Masque de warm-up calcule AVANT le calcul des features (sur les donnees + # brutes), pour ne pas dependre du fillna(0) interne a add_all_features. + valid_history_mask = _filter_warmup_period(df_monthly) + df_monthly_feat = add_all_features(df_monthly.copy()) + df_monthly_feat = df_monthly_feat[valid_history_mask.reindex(df_monthly_feat.index, fill_value=False)] + + n_dropped = int((~valid_history_mask).sum()) + if n_dropped: + logger.info(f"Warm-up : {n_dropped} lignes exclues (historique < {MIN_FEATURE_HISTORY} mois).") + daily_prices = _build_price_matrix(df_daily) daily_returns = daily_prices.pct_change().fillna(0) 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) - - portfolio_value, drifted_allocation, period_frames, rebalance_log = 100.0, {}, [], [] + + portfolio_value, benchmark_value, drifted_allocation, period_frames, rebalance_log = 100.0, 100.0, {}, [], [] monthly_dates = df_monthly_feat.index.get_level_values("date").unique().sort_values() + if len(monthly_dates) < 2: + raise ValueError( + f"Pas assez de mois exploitables après filtrage du warm-up " + f"({len(monthly_dates)} mois restants, {MIN_FEATURE_HISTORY} mois requis). " + f"Augmentez la fenêtre de backtest (BACKTEST_YEARS) ou réduisez MIN_FEATURE_HISTORY." + ) + for i, month_date in enumerate(monthly_dates[:-1]): month_data = df_monthly_feat.xs(month_date, level="date").copy() month_data["proba_upside"] = _score_with_model(model, month_data) @@ -119,11 +164,18 @@ def backtest_strategy_with_rebalancing(df_daily, df_monthly, model, benchmark_ti 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) + # Le benchmark_value retourne par _simulate_period est reinjecte a + # l'iteration suivante (au lieu d'etre code en dur a 100.0), afin que + # la courbe Benchmark compose reellement ses rendements mois apres mois. + period_df, portfolio_value, benchmark_value, drifted_allocation = _simulate_period( + allocation, drifted_allocation, trading_days, daily_returns, bench_rets, + portfolio_value, benchmark_value + ) period_frames.append(period_df) rebalance_log.append({"Date": month_date, "Allocation": allocation}) return pd.concat(period_frames), pd.DataFrame(rebalance_log).set_index("Date"), {} + def generate_live_signals(df_daily, daily_prices, model, rebalance_history): snapshot, last_date = _build_daily_snapshot(df_daily) snapshot["proba_upside"] = _score_with_model(model, snapshot) @@ -138,4 +190,4 @@ def generate_live_signals(df_daily, daily_prices, model, rebalance_history): 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 \ No newline at end of file + return out[["Ticker", "Signal", "Allocation", "Proba_Hausse"]], rebalance_history From c21ded35963e28bb6dae34792bd9e0f640606582 Mon Sep 17 00:00:00 2001 From: sora Date: Tue, 7 Jul 2026 17:32:22 +0000 Subject: [PATCH 07/13] refactor: Improve the train model --- src/models/train.py | 456 +++++++++++++++++++++++++++++--------------- 1 file changed, 305 insertions(+), 151 deletions(-) diff --git a/src/models/train.py b/src/models/train.py index 10546c5..dff107d 100644 --- a/src/models/train.py +++ b/src/models/train.py @@ -1,58 +1,170 @@ -import os +""" +Pipeline d'entraînement AlphaEdge Ensemble. + +Entraîne un modèle par marché, l'évalue (test set + walk-forward), +puis gère la promotion "champion" dans le MLflow Model Registry selon +des seuils de sécurité absolus et une comparaison relative au champion +actuel. +""" + +from __future__ import annotations + import json +import os import warnings +from dataclasses import dataclass, field from pathlib import Path -from typing import Any +from typing import Optional import pandas as pd from dotenv import load_dotenv -from sklearn.dummy import DummyClassifier -from sklearn.metrics import roc_auc_score, average_precision_score +from sklearn.metrics import average_precision_score, roc_auc_score + import mlflow -from mlflow.tracking import MlflowClient from mlflow.exceptions import MlflowException from mlflow.models.signature import infer_signature +from mlflow.tracking import MlflowClient -from const import DATA_DIR, MODEL_DIR, CONFIG_DIR, SHARPE_THRESHOLD, MAX_DD_THRESHOLD -from src.utils.metrics import calculate_financial_metrics +from const import CONFIG_DIR, DATA_DIR, MAX_DD_THRESHOLD, MODEL_DIR, SHARPE_THRESHOLD from src.features.alpha_features import add_all_features -from src.models.ensemble import AlphaEdgeEnsemble, FEATURE_GROUPS +from src.models.ensemble import AlphaEdgeEnsemble from src.utils.logger import setup_logger +from src.utils.metrics import calculate_financial_metrics load_dotenv() warnings.filterwarnings("ignore") logger = setup_logger("train") + +# ============================================================================= +# CONFIGURATION +# ============================================================================= + +MLFLOW_TRACKING_URI = "https://soradata-alphaedge-registry.hf.space" +MLFLOW_EXPERIMENT_NAME = "AlphaEdge_Ensemble_Production" +MLFLOW_USERNAME = "SORADATA" + +TEST_SET_MONTHS = 6 +MIN_TRAIN_ROWS = 100 +N_OPTUNA_TRIALS_FINAL = 50 + +WF_N_WINDOWS = 4 +WF_TEST_MONTHS = 3 +WF_N_OPTUNA_TRIALS = 20 +WF_MIN_TRAIN_ROWS = 50 +WF_MIN_TEST_ROWS = 5 + +# Tolérance de dégradation du Max Drawdown acceptée pour un challenger +# par rapport au champion actuel, avant rejet automatique. +CHALLENGER_DD_TOLERANCE = 0.02 +NO_CHAMPION_SENTINEL = -999.0 + HF_TOKEN = os.getenv("HF_TOKEN") USE_MLFLOW = bool(HF_TOKEN) if USE_MLFLOW: - os.environ["MLFLOW_TRACKING_USERNAME"] = "SORADATA" + os.environ["MLFLOW_TRACKING_USERNAME"] = MLFLOW_USERNAME os.environ["MLFLOW_TRACKING_PASSWORD"] = HF_TOKEN - mlflow.set_tracking_uri("https://soradata-alphaedge-registry.hf.space") - mlflow.set_experiment("AlphaEdge_Ensemble_Production") + mlflow.set_tracking_uri(MLFLOW_TRACKING_URI) + mlflow.set_experiment(MLFLOW_EXPERIMENT_NAME) logger.info(f"MLflow activé — tracking URI : {mlflow.get_tracking_uri()}") else: logger.warning("HF_TOKEN absent — MLflow désactivé, entraînement local uniquement.") # ============================================================================= -# HELPERS & VALIDATION +# STRUCTURES DE DONNÉES # ============================================================================= + +@dataclass +class ChampionStats: + """Métriques du modèle champion actuellement enregistré (ou sentinelles si absent).""" + sharpe: float = NO_CHAMPION_SENTINEL + sortino: float = NO_CHAMPION_SENTINEL + max_drawdown: float = NO_CHAMPION_SENTINEL + + @property + def exists(self) -> bool: + return self.sharpe != NO_CHAMPION_SENTINEL + + +@dataclass +class TrainingResult: + """Résultat consolidé d'un cycle d'entraînement pour un marché.""" + market: str + auc_test: float + apr_test: float + fin_metrics: dict + wf_auc_mean: float + mlflow_success: bool = False + mlflow_run_id: Optional[str] = None + promoted: bool = False + + def to_model_card(self) -> dict: + return { + "market": self.market, + "trained_at": pd.Timestamp.now().isoformat(), + "metrics_ml": { + "auc_test": round(self.auc_test, 4), + "apr_test": round(self.apr_test, 4), + }, + "metrics_fin": self.fin_metrics, + "walk_forward_auc_mean": round(self.wf_auc_mean, 4), + "mlflow": { + "success": self.mlflow_success, + "run_id": self.mlflow_run_id, + "promoted": self.promoted, + }, + } + + class AlphaEdgePyFunc(mlflow.pyfunc.PythonModel): - def __init__(self, model_instance): + """Wrapper PyFunc pour exposer AlphaEdgeEnsemble via l'API MLflow standard.""" + + def __init__(self, model_instance: AlphaEdgeEnsemble): self.model = model_instance - def predict(self, context, model_input, params=None): + def predict(self, context, model_input: pd.DataFrame, params: Optional[dict] = None): return self.model.predict_proba(model_input)[:, 1] + +# ============================================================================= +# CHARGEMENT & PRÉPARATION DES DONNÉES +# ============================================================================= + +def _load_market_dataset(market_name: str) -> pd.DataFrame: + """Charge le parquet mensuel d'un marché et construit la target binaire.""" + data_path = DATA_DIR / "processed" / market_name / "monthly_features.parquet" + if not data_path.exists(): + raise FileNotFoundError(f"Fichier source introuvable : {data_path}") + + df = pd.read_parquet(data_path) + price_col = "adj_close" if "adj_close" in df.columns else "adj close" + df["future_return"] = df.groupby(level="ticker")[price_col].pct_change(1).shift(-1) + df["target"] = df["future_return"].gt(0).astype(int) + return df.dropna(subset=["target", "future_return"]) + + +def _train_test_split_by_date(df: pd.DataFrame, test_months: int) -> tuple[pd.DataFrame, pd.DataFrame]: + """Split temporel (pas de shuffle) : les `test_months` derniers mois servent de test set.""" + dates = df.index.get_level_values("date") + split_date = dates.max() - pd.DateOffset(months=test_months) + df_train = add_all_features(df[dates <= split_date].copy()) + df_test = add_all_features(df[dates > split_date].copy()) + return df_train, df_test + + +# ============================================================================= +# ÉVALUATION +# ============================================================================= + def walk_forward_eval( df: pd.DataFrame, - n_windows: int = 4, - test_months: int = 3, - n_optuna_trials: int = 20, + n_windows: int = WF_N_WINDOWS, + test_months: int = WF_TEST_MONTHS, + n_optuna_trials: int = WF_N_OPTUNA_TRIALS, ) -> pd.DataFrame: - """Évaluation robuste par fenêtres glissantes pour valider la stabilité.""" + """Évaluation par fenêtres glissantes pour valider la stabilité temporelle du modèle.""" dates = df.index.get_level_values("date").unique().sort_values() results = [] @@ -67,162 +179,186 @@ def walk_forward_eval( & (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: + if len(df_tr) < WF_MIN_TRAIN_ROWS or len(df_te) < WF_MIN_TEST_ROWS or df_te["target"].nunique() < 2: + logger.debug(f"WF window {i + 1} ignorée (données insuffisantes).") 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({ + result = { "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}") + } + results.append(result) + logger.info(f"WF Window {result['window']} | AUC: {result['auc']:.4f} | APR: {result['apr']:.4f}") return pd.DataFrame(results) +def _evaluate_test_set(model: AlphaEdgeEnsemble, df_test: pd.DataFrame) -> tuple[float, float, dict]: + """Calcule AUC, APR et métriques financières sur le test set.""" + proba = model.predict_proba(df_test)[:, 1] + auc = roc_auc_score(df_test["target"], proba) + apr = average_precision_score(df_test["target"], proba) + fin_metrics = calculate_financial_metrics(df_test, probas=proba, threshold=0.5) + return auc, apr, fin_metrics + + # ============================================================================= -# PIPELINE D'ENTRAÎNEMENT +# MLFLOW : PROMOTION DU CHAMPION # ============================================================================= -def train_pipeline(market_name: str) -> tuple[AlphaEdgeEnsemble, dict]: - 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"Fichier source introuvable : {data_path}") +def _fetch_champion_stats(client: MlflowClient, registered_model_name: str, market_name: str) -> ChampionStats: + """Récupère les métriques du champion actuel, ou des sentinelles s'il n'existe pas encore.""" + try: + current_champ = client.get_model_version_by_alias(registered_model_name, "champion") + champ_metrics = client.get_run(current_champ.run_id).data.metrics + return ChampionStats( + sharpe=champ_metrics.get("Sharpe_Ratio", NO_CHAMPION_SENTINEL), + sortino=champ_metrics.get("Sortino_Ratio", NO_CHAMPION_SENTINEL), + max_drawdown=champ_metrics.get("Max_Drawdown", NO_CHAMPION_SENTINEL), + ) + except MlflowException: + logger.info(f"[{market_name}] Aucun champion trouvé. Déploiement initial.") + return ChampionStats() + + +def _should_promote(challenger_sharpe: float, challenger_sortino: float, challenger_max_dd: float, + champion: ChampionStats) -> tuple[bool, str]: + """ + Détermine si le challenger doit être promu champion. + + Règles : + 1. Sécurité absolue : Sharpe et Max Drawdown doivent dépasser les seuils configurés. + 2. Comparaison relative : le Sortino doit être strictement meilleur que celui du + champion, et le Drawdown ne doit pas se dégrader de plus de CHALLENGER_DD_TOLERANCE. + + Retourne (promu: bool, raison: str). + """ + passes_safety = (challenger_sharpe >= SHARPE_THRESHOLD) and (challenger_max_dd >= MAX_DD_THRESHOLD) + if not passes_safety: + return False, "Sharpe ou Drawdown sous les seuils de sécurité absolus" + + safer_dd = challenger_max_dd >= (champion.max_drawdown - CHALLENGER_DD_TOLERANCE) + better_sortino = challenger_sortino > champion.sortino + + if not safer_dd: + return False, "Drawdown trop dégradé par rapport au champion" + if not better_sortino: + return False, "Sortino insuffisant par rapport au champion" + return True, f"Sortino amélioré ({challenger_sortino:.2f} > {champion.sortino:.2f})" + + +def _log_and_promote_to_mlflow( + market_name: str, + model: AlphaEdgeEnsemble, + df_test: pd.DataFrame, + result: TrainingResult, +) -> None: + """Log le run MLflow, enregistre le modèle dans le Registry, et gère la promotion.""" + registered_model_name = f"AlphaEdge_Ensemble_{market_name}" + client = MlflowClient() + fin_metrics = result.fin_metrics + + try: + with mlflow.start_run(run_name=f"Ensemble_{market_name}") as run: + mlflow.log_metrics({ + "AUC_Test": result.auc_test, + "WF_AUC_Mean": result.wf_auc_mean, + "Sharpe_Ratio": fin_metrics.get("sharpe", 0.0), + "Sortino_Ratio": fin_metrics.get("sortino", 0.0), + "Calmar_Ratio": fin_metrics.get("calmar", 0.0), + "Max_Drawdown": fin_metrics.get("max_drawdown", -1.0), + }) + + 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)) + + mlflow.pyfunc.log_model( + "ensemble_model", + python_model=pyfunc_model, + signature=signature, + input_example=input_example, + ) + + result.mlflow_success = True + result.mlflow_run_id = run.info.run_id + + model_version = mlflow.register_model( + f"runs:/{run.info.run_id}/ensemble_model", registered_model_name + ) + + champion = _fetch_champion_stats(client, registered_model_name, market_name) + promote, reason = _should_promote( + challenger_sharpe=fin_metrics.get("sharpe", 0.0), + challenger_sortino=fin_metrics.get("sortino", 0.0), + challenger_max_dd=fin_metrics.get("max_drawdown", -1.0), + champion=champion, + ) + + if promote: + client.set_registered_model_alias(registered_model_name, "champion", model_version.version) + result.promoted = True + logger.info(f"[{market_name}] PROMOTION v{model_version.version} — {reason}") + else: + logger.warning(f"[{market_name}] CHALLENGER REJETÉ — {reason}") + + except MlflowException as exc: + logger.error(f"[{market_name}] Erreur durant le flux MLflow : {exc}", exc_info=True) - # 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" 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"]) - dates = df.index.get_level_values("date") - split_date = dates.max() - pd.DateOffset(months=6) +# ============================================================================= +# PIPELINE D'ENTRAÎNEMENT +# ============================================================================= - df_train = add_all_features(df[dates <= split_date].copy()) - df_test = add_all_features(df[dates > split_date].copy()) +def train_pipeline(market_name: str) -> tuple[AlphaEdgeEnsemble, dict]: + """Entraîne, évalue et (le cas échéant) promeut le modèle d'un marché donné.""" + logger.info(f"Début de l'entraînement — {market_name}") - if len(df_train) < 100: - raise ValueError(f"Volume de données insuffisant pour {market_name}: {len(df_train)} lignes.") + df = _load_market_dataset(market_name) + df_train, df_test = _train_test_split_by_date(df, TEST_SET_MONTHS) - # 2. Entraînement du modèle - model = AlphaEdgeEnsemble(n_optuna_trials=50) - model.fit(df_train, df_train["target"]) + if len(df_train) < MIN_TRAIN_ROWS: + raise ValueError(f"Volume de données insuffisant pour {market_name} : {len(df_train)} lignes.") - # 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) - - # 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) + model = AlphaEdgeEnsemble(n_optuna_trials=N_OPTUNA_TRIALS_FINAL) + model.fit(df_train, df_train["target"]) - logger.info(f"[{market_name}] Test Set -> AUC: {final_auc:.4f} | Sortino: {f_sortino:.2f} | Max DD: {f_max_dd*100:.1f}%") + final_auc, final_apr, fin_metrics = _evaluate_test_set(model, df_test) + max_dd_pct = fin_metrics.get("max_drawdown", -1.0) * 100 + logger.info( + f"[{market_name}] Test Set -> AUC: {final_auc:.4f} | " + f"Sortino: {fin_metrics.get('sortino', 0.0):.2f} | Max DD: {max_dd_pct:.1f}%" + ) - # 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_results = walk_forward_eval(df_full) 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}, - } + result = TrainingResult( + market=market_name, + auc_test=final_auc, + apr_test=final_apr, + fin_metrics=fin_metrics, + wf_auc_mean=wf_auc_mean, + ) - # 6. MLOps : MLflow & Logique de Promotion if USE_MLFLOW: - registered_model_name = f"AlphaEdge_Ensemble_{market_name}" - client = MlflowClient() + _log_and_promote_to_mlflow(market_name, model, df_test, result) - 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": f_sharpe, - "Sortino_Ratio": f_sortino, - "Calmar_Ratio": f_calmar, - "Max_Drawdown": f_max_dd - }) - - # 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)) - - 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) - - # 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") - 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é. Déploiement initial.") - - # ========================================================= - # 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 v{mv.version} ! Sortino amélioré ({f_sortino:.2f} > {champion_sortino:.2f})") - else: - 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É : Sharpe ou Drawdown sous les seuils absolus.") - - except Exception as e: - logger.error(f"[{market_name}] Erreur durant le flux MLflow : {e}") - - with open(market_model_dir / "model_card.json", "w") as f: + model_card = result.to_model_card() + with open(market_model_dir / "model_card.json", "w", encoding="utf-8") as f: json.dump(model_card, f, indent=2) return model, model_card @@ -231,26 +367,44 @@ def train_pipeline(market_name: str) -> tuple[AlphaEdgeEnsemble, dict]: # ============================================================================= # ORCHESTRATEUR # ============================================================================= -if __name__ == "__main__": + +def _load_configured_markets(config_dir: Path) -> list[str]: + """Lit les noms de marchés à partir des fichiers de configuration JSON.""" + markets = [] + for config_file in sorted(config_dir.glob("*.json")): + with open(config_file, encoding="utf-8") as f: + market_cfg = json.load(f) + market = market_cfg.get("market_name") + if market: + markets.append(market) + else: + logger.warning(f"Fichier de config sans 'market_name' ignoré : {config_file}") + return markets + + +def main() -> None: 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_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) + markets = _load_configured_markets(config_dir) + if not markets: + logger.error(f"Aucun marché configuré trouvé dans {config_dir}") + raise SystemExit(1) + + failures = [] + for market in markets: + try: + train_pipeline(market) + except Exception: + logger.critical(f"[{market}] Échec complet de l'entraînement", exc_info=True) + failures.append(market) if failures: logger.error(f"Marchés en échec : {failures}") raise SystemExit(1) + + +if __name__ == "__main__": + main() \ No newline at end of file From 9ad408b5042138e4df41280f46c3fa638016c3fb Mon Sep 17 00:00:00 2001 From: sora Date: Tue, 7 Jul 2026 17:34:39 +0000 Subject: [PATCH 08/13] refactor: Improve loader --- src/models/model_loader.py | 112 ++++++++++++++++++++++++++++++------- 1 file changed, 91 insertions(+), 21 deletions(-) diff --git a/src/models/model_loader.py b/src/models/model_loader.py index c8d3bcf..3db3306 100644 --- a/src/models/model_loader.py +++ b/src/models/model_loader.py @@ -1,12 +1,28 @@ +""" +Chargement du modèle "champion" pour l'inférence (daily run / backtest). + +Stratégie de résolution, par ordre de priorité : + 1. MLflow Model Registry (alias "champion") si HF_TOKEN est configuré. + 2. Fallback local (pickle) si MLflow est indisponible, désactivé, ou + qu'aucun alias "champion" n'existe encore. + +Un cache en mémoire évite de recharger le même modèle plusieurs fois au +sein d'un même processus (ex: backtest + génération de signaux dans le +même run quotidien). +""" + +from __future__ import annotations + import os import pickle +from functools import lru_cache from pathlib import Path -from typing import Any +from typing import Any, Optional import mlflow -from mlflow.tracking import MlflowClient -from mlflow.exceptions import MlflowException from dotenv import load_dotenv +from mlflow.exceptions import MlflowException +from mlflow.tracking import MlflowClient from const import MODEL_DIR from src.utils.logger import setup_logger @@ -14,60 +30,98 @@ load_dotenv() logger = setup_logger("model_loader") + +# ============================================================================= +# CONFIGURATION +# ============================================================================= + +MLFLOW_TRACKING_URI = "https://soradata-alphaedge-registry.hf.space" +MLFLOW_USERNAME = "SORADATA" +CHAMPION_ALIAS = "champion" +LOCAL_MODEL_FILENAME = "ensemble_model.pkl" + HF_TOKEN = os.getenv("HF_TOKEN") USE_MLFLOW = bool(HF_TOKEN) if USE_MLFLOW: - os.environ["MLFLOW_TRACKING_USERNAME"] = "SORADATA" + os.environ["MLFLOW_TRACKING_USERNAME"] = MLFLOW_USERNAME os.environ["MLFLOW_TRACKING_PASSWORD"] = HF_TOKEN - mlflow.set_tracking_uri("https://soradata-alphaedge-registry.hf.space") + mlflow.set_tracking_uri(MLFLOW_TRACKING_URI) + logger.info(f"MLflow activé pour le chargement du champion — tracking URI : {MLFLOW_TRACKING_URI}") +else: + logger.warning("HF_TOKEN absent — chargement en mode local uniquement.") + +# ============================================================================= +# CHARGEMENT DEPUIS MLFLOW +# ============================================================================= -def _load_champion_from_mlflow(market_name: str) -> Any | None: +def _load_champion_from_mlflow(market_name: str) -> Optional[Any]: """ Charge le modèle aliasé 'champion' depuis le MLflow Model Registry. - Retourne None en cas d'échec (fallback local ensuite). + Retourne None en cas d'échec (le fallback local prend alors le relais). """ registered_model_name = f"AlphaEdge_Ensemble_{market_name}" + model_uri = f"models:/{registered_model_name}@{CHAMPION_ALIAS}" try: - model_uri = f"models:/{registered_model_name}@champion" model = mlflow.pyfunc.load_model(model_uri) logger.info(f"[{market_name}] Champion chargé depuis MLflow : {model_uri}") return model - except MlflowException as e: - logger.warning(f"[{market_name}] Impossible de charger le champion MLflow : {e}") + except MlflowException as exc: + logger.warning(f"[{market_name}] Impossible de charger le champion MLflow ({model_uri}) : {exc}") return None -def _load_champion_from_local(market_name: str) -> Any | None: +# ============================================================================= +# CHARGEMENT DEPUIS LE FALLBACK LOCAL +# ============================================================================= + +def _local_model_path(market_name: str) -> Path: + return MODEL_DIR / market_name / LOCAL_MODEL_FILENAME + + +def _load_champion_from_local(market_name: str) -> Optional[Any]: """ Fallback : charge le dernier modèle sauvegardé localement (pickle). - Utilisé si MLflow est indisponible ou HF_TOKEN absent. + Utilisé si MLflow est indisponible, désactivé (HF_TOKEN absent), + ou qu'aucun alias 'champion' n'a encore été promu. """ - local_path = MODEL_DIR / market_name / "ensemble_model.pkl" + local_path = _local_model_path(market_name) if not local_path.exists(): logger.error(f"[{market_name}] Aucun modèle local trouvé à {local_path}") return None + try: with open(local_path, "rb") as f: model = pickle.load(f) logger.info(f"[{market_name}] Modèle chargé depuis le fallback local : {local_path}") return model - except Exception as e: - logger.error(f"[{market_name}] Échec du chargement local : {e}") + except (pickle.UnpicklingError, EOFError, AttributeError, ModuleNotFoundError) as exc: + # Ces erreurs signalent typiquement un fichier corrompu ou une + # incompatibilité de version entre l'environnement d'entraînement + # et celui d'inférence (classe déplacée/renommée, version sklearn...). + logger.error(f"[{market_name}] Fichier pickle illisible ou incompatible ({local_path}) : {exc}") return None +# ============================================================================= +# POINT D'ENTRÉE PUBLIC +# ============================================================================= + +@lru_cache(maxsize=None) def load_champion(market_name: str) -> Any: """ Point d'entrée unique utilisé par run_pipeline.py (daily run). + Priorité : champion MLflow -> fallback local. - Lève une exception si aucun modèle n'est disponible (le pipeline - ne doit jamais tourner sans modèle). + Le résultat est mis en cache par marché pour la durée du processus, + afin d'éviter des appels réseau MLflow redondants si le pipeline + (backtest + signaux live) charge le champion plusieurs fois. + + Lève une exception si aucun modèle n'est disponible : le pipeline + ne doit jamais tourner sans modèle. """ - model = None - if USE_MLFLOW: - model = _load_champion_from_mlflow(market_name) + model = _load_champion_from_mlflow(market_name) if USE_MLFLOW else None if model is None: model = _load_champion_from_local(market_name) @@ -78,4 +132,20 @@ def load_champion(market_name: str) -> Any: "(ni MLflow, ni local). Impossible de générer les signaux." ) - return model \ No newline at end of file + return model + + +def clear_champion_cache(market_name: Optional[str] = None) -> None: + """ + Vide le cache de load_champion. + + Utile après une nouvelle promotion (le champion vient de changer sur + MLflow) ou dans les tests, pour forcer un rechargement. + Note : lru_cache ne permet pas d'invalider une seule clé nativement, + donc on vide tout le cache quel que soit `market_name` fourni. + """ + load_champion.cache_clear() + if market_name: + logger.info(f"[{market_name}] Cache du champion invalidé.") + else: + logger.info("Cache du champion invalidé pour tous les marchés.") \ No newline at end of file From 0e010a02796dd0202b4f515a6ddaeed90d0b77af Mon Sep 17 00:00:00 2001 From: sora Date: Tue, 7 Jul 2026 17:35:18 +0000 Subject: [PATCH 09/13] feat: Add market_options dynamics --- app.py | 422 ++++++++++++++++++++++++++++++++++++--------------------- 1 file changed, 270 insertions(+), 152 deletions(-) diff --git a/app.py b/app.py index 9ec9e3f..174f908 100644 --- a/app.py +++ b/app.py @@ -1,22 +1,21 @@ -import streamlit as st +import os +import json +import time +import numpy as np import pandas as pd +import streamlit as st import plotly.express as px import plotly.graph_objects as go -from plotly.subplots import make_subplots -from datetime import datetime, timedelta from pathlib import Path -import numpy as np -import json -import yfinance as yf +from datetime import datetime, timedelta +from plotly.subplots import make_subplots from streamlit_autorefresh import st_autorefresh -import time -import os +import yfinance as yf import mlflow from mlflow.tracking import MlflowClient from mlflow.exceptions import MlflowException - # ============================================================================= # CONFIGURATION & STYLE # ============================================================================= @@ -38,8 +37,39 @@ os.environ["MLFLOW_TRACKING_USERNAME"] = "SORADATA" os.environ["MLFLOW_TRACKING_PASSWORD"] = HF_TOKEN mlflow.set_tracking_uri(MLFLOW_TRACKING_URI) + MODEL_DIR = BASE_DIR / "models" -MARKET_OPTIONS = ["CAC40", "BRVM"] + +@st.cache_data(ttl=1800, show_spinner=False) +def _discover_markets(): + """ + Decouvre dynamiquement les marches disponibles en interrogeant le repo + Hugging Face distant (dataset HF_REPO_ID), sous le prefixe data//. + Fallback sur un scan local (data/processed) si l'API HF echoue, puis sur + une liste par defaut en dernier recours. + """ + try: + from huggingface_hub import HfApi + api = HfApi() + files = api.list_repo_files(repo_id=HF_REPO_ID, repo_type="dataset", token=HF_TOKEN) + markets = sorted({ + f.split("/")[1] for f in files + if f.startswith("data/") and len(f.split("/")) > 2 + }) + if markets: + return markets + except Exception: + pass + + local_dir = BASE_DIR / "data" / "processed" + if local_dir.exists(): + found = sorted([p.name for p in local_dir.iterdir() if p.is_dir()]) + if found: + return found + + return ["CAC40", "BRVM"] + +MARKET_OPTIONS = _discover_markets() st.markdown("""