""" AHAD QUANT — RL Agent Interface (Inférence Live) — V6 UNIFIED Charge ahad_quant_unified.zip — UN seul fichier qui contient : - Le modèle PPO (réseau de neurones, poids, optimiseur) - L'ensemble supervisé (LightGBM + XGBoost + RF + TFT + TGRU + Meta) - Le scaler z-score des features Chargement : agent = RLAgent(unified_path="ahad_quant_unified.zip") signal, conf = agent.predict_signal(features, position_state) Ou depuis fichiers séparés (compatibilité V5) : agent = RLAgent(model_path="rl_agent", scaler_path="rl_scaler.pkl", ensemble_path="model_ensemble.pkl") Actions RL : 0 = HOLD/NEUTRAL 1 = LONG 2 = SHORT 3 = CLOSE """ import os import io import json import pickle import zipfile import logging import collections import numpy as np from typing import Tuple, Optional log = logging.getLogger("RLAgent") # Source unique pour le ML (voir ensemble_core.py). rl_agent.py sert à # l'INFÉRENCE LIVE et ne doit PAS dépendre de rl_env.py, qui importe # gymnasium en dur et charge tout l'environnement d'entraînement — corrige # le point #9 du diagnostic (avant, importer rl_agent.py forçait # l'installation de gymnasium même sur un déploiement live sans entraînement). import ensemble_core as _ens_core from features import NUM_FEATURES _load_ensemble = _ens_core.load_ensemble def _predict_ensemble(ensemble, raw_feat: np.ndarray, seq_buffer=None): """Wrapper local — délègue à ensemble_core (même logique exacte que rl_env.py, ahad_quant.py et backtest.py : aucune divergence possible).""" history = None if seq_buffer is not None: try: history = np.asarray(seq_buffer, dtype=np.float32) if history.ndim != 2 or len(history) == 0: history = None except Exception: history = None return _ens_core.predict_ensemble_single(ensemble, raw_feat, history=history) SEQ_LEN = _ens_core.dl_seq_len() # MAX_HOLD synchronisé sur config.py (même source unique de vérité que # rl_env.py — corrige le point #5 du diagnostic côté inférence live, pour # PositionState.to_array()). try: import config as _config _DEFAULT_MAX_HOLD = int(getattr(_config, "MAX_HOLD_CANDLES", 6)) except Exception: _DEFAULT_MAX_HOLD = 6 # ─── Constantes ────────────────────────────────────────────────────────────── ACTION_TO_SIGNAL = { 0: "neutral", 1: "long", 2: "short", 3: "neutral", } N_ML_FEATURES = NUM_FEATURES # ─── Position State ────────────────────────────────────────────────────────── class PositionState: __slots__ = ["in_position", "direction", "unrealized_pnl_pct", "candles_held", "balance_ratio"] def __init__( self, in_position: bool = False, direction: int = 0, unrealized_pnl_pct: float = 0.0, candles_held: int = 0, balance_ratio: float = 1.0, ): self.in_position = in_position self.direction = direction self.unrealized_pnl_pct = unrealized_pnl_pct self.candles_held = candles_held self.balance_ratio = balance_ratio def to_array(self, max_hold: int = _DEFAULT_MAX_HOLD) -> np.ndarray: return np.array([ float(self.in_position), float(self.direction), float(np.clip(self.unrealized_pnl_pct / 0.05, -1.0, 1.0)), float(np.clip(self.candles_held / max_hold, 0.0, 1.0)), float(np.clip(self.balance_ratio, 0.0, 2.0)), ], dtype=np.float32) # ─── RLAgent ───────────────────────────────────────────────────────────────── class RLAgent: """ Interface d'inférence pour AHAD QUANT V6 Unified. Deux modes de chargement : Mode UNIFIÉ (recommandé) : agent = RLAgent(unified_path="ahad_quant_unified.zip") Mode SÉPARÉ (compatibilité V5) : agent = RLAgent(model_path="rl_agent", scaler_path="rl_scaler.pkl", ensemble_path="model_ensemble.pkl") """ def __init__( self, # Mode unifié (repli portable) unified_path: str = None, # Mode séparé — FICHIERS VIVANTS, prioritaires (voir _load_best_available) model_path: str = None, scaler_path: str = None, ensemble_path: str = None, # Options device: str = "cpu", enabled: bool = True, ): self.enabled = enabled self._model = None self._loaded = False self._device = device self._feat_mean: Optional[np.ndarray] = None self._feat_std: Optional[np.ndarray] = None self._ensemble = None self._has_dl = False self._metadata = {} self._mtimes = {} self._seq_buffer: collections.deque = collections.deque(maxlen=SEQ_LEN) # Résout les chemins via config.py quand non fournis explicitement, # pour que les "fichiers vivants" (#2 du diagnostic) soient toujours # connus, même si l'appelant ne passe que unified_path (compat V6). import config self._model_path = model_path or getattr(config, "RL_MODEL_PATH", "rl_agent") self._scaler_path = scaler_path or getattr(config, "RL_SCALER_PATH", "rl_scaler.pkl") self._ensemble_path = ensemble_path or getattr(config, "ENSEMBLE_MODEL_PATH", "model_ensemble.pkl") self._unified_path = unified_path or getattr(config, "UNIFIED_MODEL_PATH", "ahad_quant_unified.zip") if not enabled: log.info("[RL-AGENT] Désactivé") return self._load_best_available(device) def _zip_path(self, path: str) -> str: return path if path.endswith(".zip") else path + ".zip" def _canonical_files_present(self) -> bool: """True si les fichiers vivants (rl_agent.zip + rl_scaler.pkl + model_ensemble.pkl) sont TOUS présents. Ce sont eux qui reflètent un éventuel ré-entraînement ML/RL — contrairement à ahad_quant_unified.zip, qui n'est régénéré que par un export manuel (point #2 du diagnostic).""" return ( os.path.exists(self._zip_path(self._model_path)) and os.path.exists(self._scaler_path) and os.path.exists(self._ensemble_path) ) def _load_best_available(self, device: str): """ Choisit la source à charger, par ordre de priorité : 1. Fichiers vivants (rl_agent.zip + rl_scaler.pkl + model_ensemble.pkl) s'ils sont TOUS présents — toujours prioritaires car ils reflètent le dernier ré-entraînement et permettent le hot-reload (corrige #2 et #4 du diagnostic). AVANT ce correctif, ahad_quant_unified.zip était préféré dès qu'il existait, ce qui rendait tout ré-entraînement RL inopérant en production tant que ce fichier restait sur le disque. 2. Sinon, ahad_quant_unified.zip — repli utile en déploiement portable (autre machine, sans pipeline d'entraînement local). """ if self._canonical_files_present(): log.info("[RL-AGENT] Fichiers vivants détectés — chargement séparé (prioritaire)") self._load_model(self._model_path, device) self._load_scaler(self._scaler_path) self._ensemble = _load_ensemble(self._ensemble_path) self._has_dl = self._ensemble.get("has_dl", False) if self._ensemble else False elif self._unified_path and os.path.exists(self._unified_path): log.info(f"[RL-AGENT] Fichiers vivants absents — repli sur le bundle portable : {self._unified_path}") self._load_unified(self._unified_path, device) else: log.warning("[RL-AGENT] Aucune source de modèle trouvée (ni fichiers vivants, ni bundle unifié)") self.enabled = False return self._record_mtimes() def _record_mtimes(self): self._mtimes = { "model": _ens_core.ensemble_mtime(self._zip_path(self._model_path)), "scaler": _ens_core.ensemble_mtime(self._scaler_path), "ensemble": _ens_core.ensemble_mtime(self._ensemble_path), "unified": _ens_core.ensemble_mtime(self._unified_path) if self._unified_path else 0.0, } def reload_if_stale(self) -> bool: """ Compare les mtimes actuels des fichiers source à ceux du dernier chargement (coût négligeable : quelques appels os.path.getmtime) et recharge tout (PPO + scaler + ensemble) si l'un d'eux a changé. Corrige le point #4 du diagnostic : avant, un ré-entraînement réussi en arrière-plan (auto_retrain.py) restait totalement sans effet sur le bot déjà en cours d'exécution, jusqu'à un redémarrage manuel du process. Peut être appelé à chaque itération de la boucle principale (voir unified_brain.py / ahad_quant.py::run()). Retourne True si un rechargement a effectivement eu lieu. """ if not self.enabled: return False current = { "model": _ens_core.ensemble_mtime(self._zip_path(self._model_path)), "scaler": _ens_core.ensemble_mtime(self._scaler_path), "ensemble": _ens_core.ensemble_mtime(self._ensemble_path), "unified": _ens_core.ensemble_mtime(self._unified_path) if self._unified_path else 0.0, } if current == self._mtimes: return False log.info("[RL-AGENT] Changement détecté sur les fichiers modèle — rechargement à chaud") self._seq_buffer.clear() self._load_best_available(self._device) return True # ────────────────────────────────────────────────────────────────────────── # Chargement UNIFIÉ # ────────────────────────────────────────────────────────────────────────── def _load_unified(self, path: str, device: str): """Charge tout depuis ahad_quant_unified.zip.""" log.info(f"[RL-AGENT] Chargement unifié : {path}") try: from stable_baselines3 import PPO except ImportError: log.error("[RL-AGENT] stable-baselines3 non installé.") return try: with zipfile.ZipFile(path, "r") as uz: files = uz.namelist() # ── 1. Reconstruire le zip PPO en mémoire ──────────────────── # SB3 attend un zip avec les fichiers à la racine (pas dans ppo/) ppo_buf = io.BytesIO() with zipfile.ZipFile(ppo_buf, "w", zipfile.ZIP_DEFLATED) as ppo_zip: for name in files: if name.startswith("ppo/"): inner_name = name[len("ppo/"):] # retirer le préfixe if inner_name: ppo_zip.writestr(inner_name, uz.read(name)) ppo_buf.seek(0) # ── 2. Charger le PPO depuis le BytesIO ────────────────────── self._model = PPO.load(ppo_buf, device=device) self._loaded = True log.info("[RL-AGENT] ✅ PPO chargé depuis mémoire") # ── 3. Charger le scaler ───────────────────────────────────── if "rl_scaler.pkl" in files: scaler = pickle.loads(uz.read("rl_scaler.pkl")) self._feat_mean = scaler["mean"].astype(np.float32) self._feat_std = scaler["std"].astype(np.float32) log.info("[RL-AGENT] ✅ Scaler chargé") # ── 4. Charger l'ensemble ───────────────────────────────────── if "ensemble.pkl" in files: self._ensemble = pickle.loads(uz.read("ensemble.pkl")) self._has_dl = self._ensemble.get("has_dl", False) log.info(f"[RL-AGENT] ✅ Ensemble chargé — has_dl={self._has_dl}") # ── 5. Metadata ─────────────────────────────────────────────── if "metadata.json" in files: self._metadata = json.loads(uz.read("metadata.json")) log.info(f"[RL-AGENT] Version : {self._metadata.get('version','?')}") except Exception as e: log.error(f"[RL-AGENT] Erreur chargement unifié : {e}") self.enabled = False # ────────────────────────────────────────────────────────────────────────── # Chargement SÉPARÉ (fallback) # ────────────────────────────────────────────────────────────────────────── def _load_model(self, path: str, device: str): try: from stable_baselines3 import PPO except ImportError: log.error("[RL-AGENT] stable-baselines3 non installé.") return zip_path = path if path.endswith(".zip") else path + ".zip" if not os.path.exists(zip_path): log.warning(f"[RL-AGENT] Modèle introuvable : {zip_path}") self.enabled = False return try: self._model = PPO.load(path, device=device) self._loaded = True log.info(f"[RL-AGENT] ✅ PPO chargé : {zip_path}") except Exception as e: log.error(f"[RL-AGENT] Erreur : {e}") self.enabled = False def _load_scaler(self, path: str): if not os.path.exists(path): log.warning(f"[RL-AGENT] Scaler introuvable : {path}") return try: with open(path, "rb") as f: scaler = pickle.load(f) self._feat_mean = scaler["mean"].astype(np.float32) self._feat_std = scaler["std"].astype(np.float32) log.info(f"[RL-AGENT] ✅ Scaler chargé : {path}") except Exception as e: log.error(f"[RL-AGENT] Erreur scaler : {e}") # ────────────────────────────────────────────────────────────────────────── # API Publique # ────────────────────────────────────────────────────────────────────────── def predict( self, features: np.ndarray, position_state: PositionState = None, deterministic: bool = True, ) -> int: """Retourne l'action RL (0=HOLD, 1=LONG, 2=SHORT, 3=CLOSE).""" if not self.enabled or not self._loaded or self._model is None: return 0 obs = self._build_obs(features, position_state) try: action, _ = self._model.predict(obs, deterministic=deterministic) # numpy 2.x : int() ne fonctionne plus sur les arrays 1-D → .flat[0] return int(action.flat[0]) if hasattr(action, "flat") else int(action) except Exception as e: log.error(f"[RL-AGENT] Erreur predict : {e}") return 0 def predict_signal( self, features: np.ndarray, position_state: PositionState = None, return_action: bool = False, ): """Raccourci : features → (signal, confidence), ou (signal, confidence, action) si return_action=True. L'action brute PPO (0=HOLD,1=LONG,2=SHORT,3=CLOSE) est nécessaire pour le buffer d'apprentissage continu — voir filter_signal().""" action = self.predict(features, position_state) signal = ACTION_TO_SIGNAL.get(action, "neutral") # Confidence basée sur la certitude de l'ensemble intégré _, ens_conf = _predict_ensemble( self._ensemble, features.astype(np.float32), self._seq_buffer if self._has_dl else None, ) # Confidence finale = moyenne ens_conf + base fixe confidence = round(0.70 + ens_conf * 0.25, 4) # [0.70, 0.95] if return_action: return signal, confidence, action return signal, confidence def filter_signal( self, ml_signal: str, ml_confidence: float, features: np.ndarray, position_state: "PositionState" = None, ): """ Filtre le signal ML via le PPO. - RL confirme ML → boost confiance - RL override fort → suit le RL - RL neutral / erreur → fallback sur ML (ne bloque pas) Retourne (signal, confidence, rl_action) — rl_action est l'action PPO brute (None si une exception interne a empêché toute inférence RL). Corrige le point #8 du diagnostic : avant, cette action réelle n'était jamais propagée jusqu'au contexte d'ouverture de trade, et online_learner.py devait la DEVINER à partir du signal final (1 si LONG, 2 sinon), cassant la fidélité du replay RL. """ try: rl_signal, rl_confidence, rl_action = self.predict_signal( features, position_state, return_action=True ) except Exception: # Erreur interne RL → on laisse passer le signal ML return ml_signal, ml_confidence, None # Seuils lus depuis config.py (et non os.getenv directement) pour # que la calibration adaptative (online_learner.calibrate_threshold, # qui peut ajuster config.RL_OVERRIDE_THRESHOLD en mémoire) ait # réellement un effet — corrige le point #7 du diagnostic. import config as _cfg # RL confirme la direction ML if rl_signal == ml_signal: boost = float(getattr(_cfg, "RL_CONFIDENCE_BOOST", 0.05)) return ml_signal, round(min(ml_confidence + boost, 0.99), 4), rl_action # RL override avec haute confiance override_thr = float(getattr(_cfg, "RL_OVERRIDE_THRESHOLD", 0.82)) if rl_signal != "neutral" and rl_confidence >= override_thr: return rl_signal, rl_confidence, rl_action # RL dit neutral ou pas assez confiant → on garde le signal ML return ml_signal, ml_confidence, rl_action def reset_buffer(self): """Vide le buffer de séquence (appeler entre paires / sessions).""" self._seq_buffer.clear() def info(self) -> dict: """Retourne les métadonnées du modèle chargé.""" return { **self._metadata, "ppo_loaded": self._loaded, "ensemble": self._ensemble is not None, "has_dl": self._has_dl, "scaler": self._feat_mean is not None, } def is_ready(self) -> bool: return self.enabled and self._loaded and self._model is not None # ────────────────────────────────────────────────────────────────────────── # Interne # ────────────────────────────────────────────────────────────────────────── def _build_obs( self, raw_features: np.ndarray, position_state: Optional[PositionState], ) -> np.ndarray: """Construit le vecteur d'observation 69 dims (identique à rl_env).""" feat = raw_features.astype(np.float32).copy() if len(feat) > N_ML_FEATURES: feat = feat[:N_ML_FEATURES] elif len(feat) < N_ML_FEATURES: feat = np.pad(feat, (0, N_ML_FEATURES - len(feat))) if self._feat_mean is not None and self._feat_std is not None: feat = (feat - self._feat_mean) / self._feat_std feat = np.nan_to_num(feat, nan=0.0, posinf=0.0, neginf=0.0) if position_state is None: position_state = PositionState() pos = position_state.to_array() if self._ensemble is not None: self._seq_buffer.append(raw_features.astype(np.float32)) ens_proba, ens_conf = _predict_ensemble( self._ensemble, raw_features.astype(np.float32), self._seq_buffer if self._has_dl else None, ) ens_feats = np.array([ens_proba, ens_conf], dtype=np.float32) obs = np.concatenate([feat, pos, ens_feats]) else: obs = np.concatenate([feat, pos]) return obs[np.newaxis, :].astype(np.float32) # ─── Singleton global ───────────────────────────────────────────────────────── _agent_instance: Optional[RLAgent] = None def get_rl_agent() -> RLAgent: global _agent_instance if _agent_instance is None: import config unified = getattr(config, "UNIFIED_MODEL_PATH", "ahad_quant_unified.zip") _agent_instance = RLAgent( unified_path = unified, enabled = getattr(config, "USE_RL_AGENT", False), ) return _agent_instance def reset_rl_agent(): global _agent_instance _agent_instance = None log.info("[RL-AGENT] Singleton réinitialisé")