""" AHAD QUANT — Online Learner (Apprentissage Continu V7) ========================================================= Apprentissage continu à partir des expériences stockées dans le buffer. 3 mécanismes distincts : 1. LightGBM WARM-START (quotidien, ~2-5 min, local CPU) - Récupère les trades LOSS + TIMEOUT des dernières 24h - Pondère chaque sample par error_weight - Ré-entraîne LightGBM avec init_model=model_actuel (50 arbres supplémentaires) - Remplace model.pkl SEULEMENT si accuracy >= accuracy_actuelle - 0.5% 2. RL EXPERIENCE REPLAY (hebdomadaire, ~2-4h) - Convertit les trades réels du buffer en épisodes RL (obs, action, reward, next_obs) - Injecte ces épisodes dans le fine-tune PPO SB3 3. CALIBRATION SEUIL (immédiat, après chaque trade fermé) - Sliding window sur les 50 derniers trades - Si win_rate < 50% → MIN_CONFIDENCE += 0.01 - Si win_rate > 65% → MIN_CONFIDENCE -= 0.005 - Borné entre [0.62, 0.88] - Sauvegardé dans adaptive_state.json Usage : from experience_buffer import get_experience_buffer from online_learner import OnlineLearner buf = get_experience_buffer() learner = OnlineLearner(buf) learner.daily_update() new_conf = learner.calibrate_threshold(recent_n=50) episodes = learner.prepare_rl_episodes() """ import json import logging import os import pickle import shutil import time from datetime import datetime, timezone from typing import Dict, List, Optional, Tuple import numpy as np import config from experience_buffer import ExperienceBuffer import ensemble_core as _ens_core from features import NUM_FEATURES log = logging.getLogger("OnlineLearner") # ── Constantes ──────────────────────────────────────────────────────────────── WARMSTART_ROUNDS = 50 # arbres supplémentaires par cycle quotidien WARMSTART_MIN_TRADES = 20 # minimum de trades échoués pour déclencher CALIBRATION_WINDOW = 50 # trades pour sliding win rate CONFIDENCE_MIN = 0.62 # plancher MIN_CONFIDENCE CONFIDENCE_MAX = 0.88 # plafond MIN_CONFIDENCE CONFIDENCE_STEP_UP = 0.01 # hausse si win_rate < 50% CONFIDENCE_STEP_DOWN = 0.005 # baisse si win_rate > 65% ACCURACY_TOLERANCE = 0.005 # -0.5% toléré pour accepter le nouveau modèle ADAPTIVE_STATE_FILE = "adaptive_state.json" # Calibration jointe ML/RL (corrige le point #7 du diagnostic) : avant, # config.RL_OVERRIDE_THRESHOLD (utilisé par rl_agent.py::filter_signal pour # décider si le RL peut s'imposer face au ML) n'était JAMAIS ajusté par # calibrate_threshold(), qui ne touchait que MIN_CONFIDENCE — les deux # seuils dérivaient l'un de l'autre silencieusement avec le temps. On les # fait maintenant bouger ensemble, à écart ("marge") constant. RL_OVERRIDE_MIN = 0.65 RL_OVERRIDE_MAX = 0.95 class OnlineLearner: """ Apprentissage incrémental à partir des expériences réelles de trading. """ def __init__( self, buffer: ExperienceBuffer, model_path: Optional[str] = None, ensemble_path: Optional[str] = None, adaptive_state_file: str = ADAPTIVE_STATE_FILE, ): self.buffer = buffer self.model_path = model_path or getattr(config, "MODEL_PATH", "model.pkl") self.ensemble_path = ensemble_path or getattr(config, "ENSEMBLE_MODEL_PATH", "model_ensemble.pkl") self.state_file = adaptive_state_file self._state = self._load_adaptive_state() # ── Adaptive state I/O ─────────────────────────────────────────────────── def _load_adaptive_state(self) -> Dict: default_margin = round( getattr(config, "RL_OVERRIDE_THRESHOLD", 0.82) - getattr(config, "MIN_CONFIDENCE", 0.72), 4 ) defaults = { "min_confidence": getattr(config, "MIN_CONFIDENCE", 0.72), "rl_override_threshold": getattr(config, "RL_OVERRIDE_THRESHOLD", 0.82), "rl_override_margin": default_margin, "last_calibration": None, "win_rate_50": 0.0, "avg_pnl_50": 0.0, "total_trades": 0, "emergency_retrain_at": None, "confidence_history": [], } if not os.path.exists(self.state_file): return defaults try: with open(self.state_file, encoding="utf-8") as f: data = json.load(f) # Fusionner avec les defaults (robustesse) return {**defaults, **data} except Exception as e: log.warning(f"Erreur chargement adaptive_state : {e} — reset") return defaults def _save_adaptive_state(self) -> None: try: self._state["last_calibration"] = datetime.now(timezone.utc).isoformat() self._state["total_trades"] = self.buffer.total_trades tmp = self.state_file + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(self._state, f, indent=2, ensure_ascii=False) os.replace(tmp, self.state_file) except Exception as e: log.error(f"Erreur sauvegarde adaptive_state : {e}") @property def current_confidence(self) -> float: return self._state.get("min_confidence", getattr(config, "MIN_CONFIDENCE", 0.72)) # ── Mécanisme 3 : Calibration du seuil de confiance ───────────────────── def calibrate_threshold(self, recent_n: int = CALIBRATION_WINDOW) -> float: """ Ajuste MIN_CONFIDENCE dynamiquement selon le win_rate récent. - win_rate < 50% → hausse le seuil (devient plus sélectif) - win_rate > 65% → baisse le seuil (devient plus actif) Retourne la nouvelle valeur de MIN_CONFIDENCE. """ if not getattr(config, "CONTINUOUS_LEARNING_ENABLED", True): return self.current_confidence recent = self.buffer.get_last_n(recent_n) if len(recent) < 5: return self.current_confidence # pas assez de données wins = sum(1 for e in recent if e.get("outcome") == "WIN") win_rate = wins / len(recent) current_conf = self.current_confidence if win_rate < 0.50: new_conf = min(current_conf + CONFIDENCE_STEP_UP, CONFIDENCE_MAX) reason = f"win_rate={win_rate:.1%} < 50% → hausse seuil" elif win_rate > 0.65: new_conf = max(current_conf - CONFIDENCE_STEP_DOWN, CONFIDENCE_MIN) reason = f"win_rate={win_rate:.1%} > 65% → baisse seuil" else: new_conf = current_conf reason = f"win_rate={win_rate:.1%} stable — pas de changement" changed = abs(new_conf - current_conf) > 1e-5 self._state["min_confidence"] = round(new_conf, 4) self._state["win_rate_50"] = round(win_rate, 4) # Historique des 10 dernières valeurs history = self._state.get("confidence_history", []) history.append(round(new_conf, 4)) self._state["confidence_history"] = history[-10:] # Synchronise config en mémoire (pour la session courante) config.MIN_CONFIDENCE = new_conf # ── Calibration jointe RL (corrige le point #7) ────────────────── # RL_OVERRIDE_THRESHOLD suit MIN_CONFIDENCE à marge constante (l'écart # initial entre les deux, persisté), au lieu de rester figé pendant # que MIN_CONFIDENCE dérive. rl_agent.py::filter_signal() lit déjà # config.RL_OVERRIDE_THRESHOLD (et non plus une variable d'env figée), # donc cet ajustement a un effet réel dès le prochain signal. margin = self._state.get( "rl_override_margin", round(getattr(config, "RL_OVERRIDE_THRESHOLD", 0.82) - current_conf, 4), ) new_rl_override = float(np.clip(new_conf + margin, RL_OVERRIDE_MIN, RL_OVERRIDE_MAX)) config.RL_OVERRIDE_THRESHOLD = round(new_rl_override, 4) self._state["rl_override_threshold"] = round(new_rl_override, 4) self._save_adaptive_state() if changed: log.info( f"[CALIBRATION] {reason} | conf {current_conf:.3f} → {new_conf:.3f} " f"| RL_OVERRIDE_THRESHOLD → {new_rl_override:.3f}" ) return new_conf # ── Mécanisme 1 : LightGBM Warm-Start quotidien ────────────────────────── def daily_update( self, loss_trades: Optional[List[Dict]] = None, hours: float = 24.0, ) -> bool: """ Warm-start LightGBM sur les trades LOSS + TIMEOUT des dernières h. Si USE_ENSEMBLE=true (configuration par défaut) ET que model_ensemble.pkl existe, le warm-start cible le sous-modèle `lgbm` EMBARQUÉ DANS L'ENSEMBLE, et l'acceptation est évaluée au niveau de l'ensemble complet (ensemble_core.predict_ensemble_batch) plutôt que sur le seul LightGBM isolé. AVANT ce correctif, cette méthode ne mettait à jour QUE model.pkl — un fichier jamais relu en production dès que USE_ENSEMBLE=true (la valeur par défaut), ce qui rendait ce mécanisme un no-op silencieux pour la quasi-totalité des déploiements (point #3 du diagnostic). Repli sur model.pkl (LightGBM autonome) si USE_ENSEMBLE=false ou si model_ensemble.pkl est absent — comportement identique à avant pour ce cas. Retourne True si le modèle a été mis à jour. """ if not getattr(config, "WARMSTART_ENABLED", True): log.debug("[WARMSTART] Désactivé (WARMSTART_ENABLED=false)") return False use_ensemble = bool(getattr(config, "USE_ENSEMBLE", True)) and os.path.exists(self.ensemble_path) target_path = self.ensemble_path if use_ensemble else self.model_path if not os.path.exists(target_path): log.warning(f"[WARMSTART] Modèle introuvable ({target_path}) — skip") return False # ── Charger : préférer le candidat accumulé si disponible ─────────── # [FIX] Si le warmstart précédent a été rejeté (accuracy insuffisante), # les arbres qu'il a appris sont sauvegardés dans .candidate. # On repart de ce candidat plutôt que de la production pour que les # mauvais trades s'accumulent cycle après cycle, même quand le modèle # actif reste inchangé. candidate_path = target_path + ".candidate" start_model_path = candidate_path if os.path.exists(candidate_path) else target_path start_label = "candidate" if os.path.exists(candidate_path) else "production" try: with open(start_model_path, "rb") as f: loaded = pickle.load(f) except Exception as e: log.error(f"[WARMSTART] Erreur chargement modèle ({start_model_path}) : {e}") return False log.info(f"[WARMSTART] Point de départ : {start_label} ({start_model_path})") if use_ensemble: ensemble_dict = loaded current_model = ensemble_dict.get("lgbm") if current_model is None: log.warning("[WARMSTART] Pas de sous-modèle LightGBM dans l'ensemble — skip") return False else: ensemble_dict = None current_model = loaded["model"] if isinstance(loaded, dict) else loaded # Récupérer les trades échoués if loss_trades is None: loss_trades = self.buffer.get_recent(hours=hours, outcome_filter=["LOSS", "TIMEOUT"]) min_trades = getattr(config, "WARMSTART_MIN_TRADES", WARMSTART_MIN_TRADES) if len(loss_trades) < min_trades: log.info(f"[WARMSTART] Seulement {len(loss_trades)} trades échoués " f"(min={min_trades}) — skip") return False try: import lightgbm as lgb except ImportError: log.error("[WARMSTART] LightGBM non installé — skip") return False log.info( f"[WARMSTART] Démarrage warm-start ({'ensemble' if use_ensemble else 'LightGBM seul'}) " f"sur {len(loss_trades)} trades échoués..." ) # Construire le dataset X, y, weights = _build_dataset_from_trades(loss_trades) if X is None or len(X) < 10: log.warning("[WARMSTART] Dataset insuffisant — skip") return False # Préparer le dataset LightGBM avec pondération par error_weight train_data = lgb.Dataset(X, label=y, weight=weights) # Params hérités du modèle courant si disponibles params = { "objective": "binary", "metric": "binary_logloss", "learning_rate": 0.03, "num_leaves": 63, "feature_fraction": 0.8, "bagging_fraction": 0.8, "bagging_freq": 5, "min_child_samples": 10, "verbose": -1, } try: new_model = lgb.train( params, train_data, num_boost_round = WARMSTART_ROUNDS, init_model = current_model, # ← warm-start cumulatif valid_sets = [train_data], # requis par early_stopping (LightGBM >= 4.x) callbacks = [lgb.early_stopping(10, verbose=False), lgb.log_evaluation(period=-1)], ) except Exception as e: log.error(f"[WARMSTART] Erreur entraînement : {e}") return False # ── Évaluation de l'acceptation ─────────────────────────────────── if use_ensemble and ensemble_dict.get("meta") is not None and ensemble_dict.get("scaler") is not None: # Juger l'effet RÉEL sur la prédiction finale de l'ensemble (et # non sur le seul sous-modèle LightGBM isolé) : on construit un # ensemble candidat avec le lgbm mis à jour, et on compare ses # probabilités à celles de l'ensemble actuel sur le même jeu de # trades. Délègue à ensemble_core — la même fonction utilisée # partout ailleurs, aucune divergence possible. current_probas = _ens_core.predict_ensemble_batch(ensemble_dict, X)[:, 0] current_acc = float(((current_probas > 0.5) == y).mean()) candidate_ensemble = dict(ensemble_dict) candidate_ensemble["lgbm"] = new_model new_probas = _ens_core.predict_ensemble_batch(candidate_ensemble, X)[:, 0] new_acc = float(((new_probas > 0.5) == y).mean()) else: current_preds = current_model.predict(X) > 0.5 current_acc = float((current_preds == y).mean()) new_preds = new_model.predict(X) > 0.5 new_acc = float((new_preds == y).mean()) # Accepter si accuracy >= actuelle - tolérance # IMPORTANT : on compare toujours par rapport à la PRODUCTION (target_path), # pas par rapport au candidat — c'est la production qui est le champion. prod_acc = current_acc # current_acc vient du modèle chargé (candidat ou prod) if os.path.exists(candidate_path): # Re-évaluer l'accuracy de la PRODUCTION pour comparaison juste try: with open(target_path, "rb") as _pf: prod_loaded = pickle.load(_pf) if use_ensemble: prod_ens = prod_loaded prod_lgbm = prod_ens.get("lgbm") if prod_lgbm is not None: prod_probas = _ens_core.predict_ensemble_batch(prod_ens, X)[:, 0] prod_acc = float(((prod_probas > 0.5) == y).mean()) else: prod_m = prod_loaded["model"] if isinstance(prod_loaded, dict) else prod_loaded prod_acc = float(((prod_m.predict(X) > 0.5) == y).mean()) except Exception: pass # si lecture prod échoue, on garde current_acc comme référence min_acceptable = prod_acc - ACCURACY_TOLERANCE if new_acc >= min_acceptable: # ✅ Nouveau modèle meilleur → remplace la production backup = target_path + ".warmstart_backup" shutil.copy2(target_path, backup) if use_ensemble: ensemble_dict["lgbm"] = new_model with open(target_path, "wb") as _f: pickle.dump(ensemble_dict, _f) else: with open(target_path, "wb") as _f: pickle.dump(new_model, _f) # Candidat obsolète — nettoyer if os.path.exists(candidate_path): os.remove(candidate_path) log.info("[WARMSTART] Candidat accumulé promu en production — fichier .candidate supprimé") log.info( f"[WARMSTART] ✅ Modèle mis à jour ({target_path}) : " f"acc {prod_acc:.3f} → {new_acc:.3f} ({new_acc - prod_acc:+.3f})" ) return True else: # ❌ Pas encore assez bon pour remplacer la production # [FIX] On sauvegarde quand même les arbres appris en tant que # candidat. Le prochain cycle repartira de ce candidat et continuera # d'absorber les mauvais trades — l'apprentissage ne s'arrête jamais. try: if use_ensemble: candidate_dict = dict(ensemble_dict) candidate_dict["lgbm"] = new_model with open(candidate_path, "wb") as _cf: pickle.dump(candidate_dict, _cf) else: with open(candidate_path, "wb") as _cf: pickle.dump(new_model, _cf) log.info( f"[WARMSTART] ⏳ Candidat sauvegardé ({candidate_path}) : " f"acc {new_acc:.3f} vs prod {prod_acc:.3f} — " f"le prochain cycle repartira de ce candidat" ) except Exception as _ce: log.warning(f"[WARMSTART] Impossible de sauvegarder le candidat : {_ce}") log.warning( f"[WARMSTART] ❌ Nouveau modèle rejeté (prod conservée) : " f"acc {new_acc:.3f} < {min_acceptable:.3f}" ) return False # ── Mécanisme 2 : Préparation des épisodes RL pour injection PPO ───────── def prepare_rl_episodes( self, buffer: Optional[ExperienceBuffer] = None, min_trades: int = 50, ) -> List[Dict]: """ Convertit les trades réels du buffer en épisodes RL. Format : {"obs": [...67 dims], "action": int, "reward": float, "done": bool} Ces épisodes seront injectés dans le replay buffer SB3 lors du fine-tune PPO. Retourne une liste d'épisodes RL. """ buf = buffer or self.buffer if not getattr(config, "RL_REAL_REPLAY_ENABLED", True): log.debug("[RL-REPLAY] Désactivé (RL_REAL_REPLAY_ENABLED=false)") return [] if buf.total_trades < min_trades: log.info(f"[RL-REPLAY] Seulement {buf.total_trades} trades (min={min_trades}) — skip") return [] # Cibler en priorité les LOSS et TIMEOUT trades = buf.get_loss_trades(min_weight=1.5) if len(trades) < 10: trades = buf.get_all() episodes = [] for trade in trades: try: episode = _trade_to_rl_episode(trade) if episode: episodes.append(episode) except Exception as e: log.debug(f"[RL-REPLAY] Skipping trade {trade.get('id','?')} : {e}") log.info(f"[RL-REPLAY] {len(episodes)} épisodes RL préparés depuis {len(trades)} trades") return episodes # ── Rapport ────────────────────────────────────────────────────────────── def get_status(self) -> Dict: """Retourne l'état actuel du learner.""" return { "min_confidence": self.current_confidence, "total_trades": self.buffer.total_trades, "last_calibration": self._state.get("last_calibration"), "win_rate_50": self._state.get("win_rate_50", 0.0), "buffer_stats": self.buffer.stats(window=50), } # ── Helpers privés ──────────────────────────────────────────────────────────── def _build_dataset_from_trades( trades: List[Dict], ) -> Tuple[Optional[np.ndarray], Optional[np.ndarray], Optional[np.ndarray]]: """ Construit (X, y, weights) depuis une liste d'expériences. X = features (62-dim), y = label (WIN=1, LOSS/TIMEOUT=0), weights = error_weight. """ X_list, y_list, w_list = [], [], [] for trade in trades: features = trade.get("features") if not features or len(features) != NUM_FEATURES: continue outcome = trade.get("outcome", "WIN") label = 1.0 if outcome == "WIN" else 0.0 weight = float(trade.get("error_weight", 1.0)) X_list.append(features) y_list.append(label) w_list.append(weight) if len(X_list) < 5: return None, None, None return ( np.array(X_list, dtype=np.float32), np.array(y_list, dtype=np.float32), np.array(w_list, dtype=np.float32), ) def _trade_to_rl_episode(trade: Dict) -> Optional[Dict]: """ Convertit un trade réel en épisode RL au format SB3. obs = [features(62), in_position, direction, unrealized_pnl, candles_held, balance_ratio, ens_proba_long, ens_confidence] → 69 dims. [BUG CORRIGÉ 21/06/2026] Cette fonction ne produisait avant que 67 dims (sans les 2 dims "ensemble"), alors que le modèle PPO réellement entraîné et déployé attend 69 dims dans TOUS les cas : voir rl_env.py::_get_obs() (obs = feat + pos_feats + ens_feats quand self._ensemble est actif, ce qui est la config par défaut) et rl_agent.py::predict() qui construit l'obs de production de façon identique. Avec seulement 67 dims, _build_real_trade_buffer() (rl_train.py) aurait filtré TOUS les trades réels pour incompatibilité de dimensions — silencieusement, comme les bugs précédents. Les 2 dims manquantes [ens_proba_long, ens_confidence] sont reconstruites EXACTEMENT (pas une approximation) à partir des champs déjà stockés dans chaque trade — confidence et signal — en inversant la formule utilisée partout ailleurs (ensemble_core.py::predict_ensemble_batch) : confidence = |proba_long - 0.5| * 2 donc : proba_long = 0.5 + confidence/2 si signal == LONG (proba_long >= 0.5) proba_long = 0.5 - confidence/2 si signal == SHORT (proba_long < 0.5) """ features = trade.get("features") if not features or len(features) != NUM_FEATURES: return None signal = trade.get("signal", "LONG").upper() pnl = float(trade.get("pnl", 0.0)) hold = int(trade.get("hold_candles", 1)) outcome = trade.get("outcome", "WIN") # Reconstituer l'observation 69-dim in_position = 1.0 direction = 1.0 if signal == "LONG" else -1.0 unrealized_pnl = float(np.clip(pnl, -1.0, 1.0)) candles_norm = float(np.clip(hold / 24.0, 0.0, 1.0)) balance_ratio = 1.0 + float(np.clip(pnl * 10, -0.5, 0.5)) # Dims ensemble [proba_long, confidence] — reconstruction exacte (voir # docstring ci-dessus). confidence par défaut à 0.5 si absent (donne # proba_long=0.75/0.25 selon le sens — meilleur repli que de planter). confidence = float(trade.get("confidence", 0.5)) ens_confidence = float(np.clip(confidence, 0.0, 1.0)) if signal == "LONG": ens_proba_long = 0.5 + ens_confidence / 2.0 else: ens_proba_long = 0.5 - ens_confidence / 2.0 obs = ( list(features) + [in_position, direction, unrealized_pnl, candles_norm, balance_ratio] + [ens_proba_long, ens_confidence] ) # Action RL originale (si disponible), sinon déduite rl_action = trade.get("rl_action") if rl_action is None: rl_action = 1 if signal == "LONG" else 2 # LONG=1, SHORT=2 # Récompense : basée sur le résultat réel, clippée comme dans l'entraînement if outcome == "WIN": reward = float(np.clip(pnl * 10, 0.1, 1.0)) elif outcome == "LOSS": reward = float(np.clip(pnl * 10, -1.0, -0.1)) else: # TIMEOUT reward = float(np.clip(pnl * 5, -0.5, 0.5)) return { "obs": obs, "action": int(rl_action), "reward": float(np.clip(reward, -1.0, 1.0)), "done": True, # chaque trade = fin d'épisode "trade_id": trade.get("id", ""), } # ── CLI de test ─────────────────────────────────────────────────────────────── if __name__ == "__main__": import logging logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(name)s] %(message)s") from experience_buffer import ExperienceBuffer # Créer un buffer de test buf = ExperienceBuffer(max_size=200, buffer_file="test_learner_buffer.json") for i in range(60): outcome = ["WIN", "LOSS", "TIMEOUT"][i % 3] buf.add({ "pair": "EURUSD", "signal": "LONG", "confidence": 0.75, "ml_probas": [0.72, 0.73, 0.71, 0.74, 0.72], "rl_action": 1, "rl_agreed": True, "features": list(np.random.randn(NUM_FEATURES).astype(float)), "regime": "NORMAL", "entry_price": 1.0850, "exit_price": 1.0880 if outcome == "WIN" else 1.0820, "pnl": 0.03 if outcome == "WIN" else -0.02, "outcome": outcome, "hold_candles": 4, }) learner = OnlineLearner(buf, adaptive_state_file="test_adaptive_state.json") # Test calibration new_conf = learner.calibrate_threshold(recent_n=50) print(f"\nNouvelle confidence : {new_conf:.4f}") # Test préparation épisodes RL episodes = learner.prepare_rl_episodes(min_trades=10) print(f"Épisodes RL préparés : {len(episodes)}") # Statut print(f"\nStatut learner : {json.dumps(learner.get_status(), indent=2)}") # Nettoyage import os for f in ["test_learner_buffer.json", "test_adaptive_state.json"]: if os.path.exists(f): os.remove(f) print("\n✅ OnlineLearner — test OK")