""" AHAD QUANT — Gymnasium Trading Environment (V6 — UNIFIED + ML CACHE) Environnement Forex pour l entraînement de l agent RL (PPO). PATCH PERFORMANCE : Les prédictions ensemble (LGB+XGB+RF) sont pré-calculées en batch au chargement des données → lookup O(1) pendant l entraînement. Gain attendu : 15 fps → 500-1500 fps. Observation (69 dims) : [0:62] → 62 features techniques normalisées (identiques à train.py) [62] → in_position (0 ou 1) [63] → direction (-1=SHORT, 0=flat, 1=LONG) [64] → unrealized_pnl (normalisé [-1, 1]) [65] → candles_held (normalisé [0, 1]) [66] → balance_ratio (balance courante / balance initiale, clampé [0, 2]) [67] → ensemble_proba (proba LONG du méta-modèle, [0, 1]) [68] → ensemble_confidence (certitude de l ensemble, [0, 1]) Actions (Discrete 4) : 0 → HOLD (ne rien faire) 1 → LONG (ouvrir long, ou fermer short + ouvrir long) 2 → SHORT (ouvrir short, ou fermer long + ouvrir short) 3 → CLOSE (fermer la position courante) """ import os import json import pickle import random import logging import collections import numpy as np log = logging.getLogger("RL-ENV") try: import gymnasium as gym from gymnasium import spaces except ImportError: raise ImportError( "gymnasium non installé. Exécuter : pip install gymnasium>=0.29.0" ) from features import build_features, FEATURE_NAMES, NUM_FEATURES from rl_reward import RewardTracker, compute_step_reward, compute_episode_metrics import ensemble_core as _ens_core # ─── Constantes ────────────────────────────────────────────────────────────── ENS_EXTRA = 2 OBS_DIM = NUM_FEATURES + 5 + ENS_EXTRA # 62 + 5 + 2 = 69 N_ACTIONS = 4 FEE_RATE = 0.00004 MIN_CANDLES = 300 WARMUP = 50 # MAX_HOLD synchronisé sur config.MAX_HOLD_CANDLES (source unique de vérité, # utilisée aussi par risk_manager.py / ahad_quant.py pour fermer les positions # en live). AVANT ce correctif, cette constante était figée à 6 ici alors que # config.py vaut 3 par défaut : l'agent RL était entraîné en croyant pouvoir # garder une position deux fois plus longtemps qu'en réalité (vrai écart # train/live). NOTE IMPORTANTE : changer cette constante seule NE CORRIGE PAS # rétroactivement les poids déjà entraînés de rl_agent.zip / rl_checkpoints — # un ré-entraînement via rl_train.py est nécessaire pour que l'agent en # bénéficie pleinement. Le fallback à 6 n'est utilisé que si config.py est # introuvable (ne devrait jamais arriver en usage normal). try: import config as _config MAX_HOLD = int(getattr(_config, "MAX_HOLD_CANDLES", 6)) except Exception: MAX_HOLD = 6 SEQ_LEN = 168 EP_MIN_LEN = 500 EP_MAX_LEN = 2000 # ─── Chargement / prédiction ensemble ──────────────────────────────────────── # Délégué intégralement à ensemble_core.py — SOURCE UNIQUE pour le ML, partagée # avec ahad_quant.py (live), backtest.py (validation) et rl_agent.py (inférence # RL live). Les alias ci-dessous préservent la compatibilité avec le code # existant de ce fichier (qui appelle _load_ensemble / _predict_ensemble_batch # par leur nom historique) sans dupliquer la logique. _load_ensemble = _ens_core.load_ensemble _predict_ensemble_batch = _ens_core.predict_ensemble_batch def _predict_ensemble(ensemble, raw_feat: np.ndarray, seq_buffer=None): """ Prédit pour UNE SEULE observation — utilisé par rl_agent.py en inférence live (à chaque nouvelle bougie, paire par paire), contrairement à _predict_ensemble_batch() qui traite tout un historique d'un coup au chargement. Délègue à ensemble_core.predict_ensemble_single(), qui réutilise EN INTERNE la même fonction de stacking que le batch d'entraînement → l'inférence live et l'entraînement appliquent EXACTEMENT la même logique ML, y compris pour les modèles séquentiels (TFT/TransformerGRU) quand has_dl=True. raw_feat : vecteur (NUM_FEATURES,) de features brutes (non normalisées) seq_buffer : deque/array de l'historique récent des features (pour les modèles séquentiels TFT/TGRU). Si fourni et que l'ensemble a has_dl=True, il est converti en fenêtre numpy et transmis à ensemble_core ; sinon, les composantes DL sont simplement omises du stacking (comme tout sous-modèle absent), sans jamais lever d'exception. Retourne (proba, confidence) — deux floats Python (pas un array). """ 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) # ─── Environnement ──────────────────────────────────────────────────────────── class AhadQuantForexEnv(gym.Env): metadata = {"render_modes": ["human"]} def __init__( self, data_dir: str = "data", pairs: list = None, sl_pct: float = 0.005, tp_pct: float = 0.0075, leverage: float = 30.0, risk_per_trade: float = 0.01, initial_balance: float = 100.0, normalize_obs: bool = True, ensemble_path: str = None, seed: int = None, ): super().__init__() self.data_dir = data_dir self.sl_pct = sl_pct self.tp_pct = tp_pct self.leverage = leverage self.risk_per_trade = risk_per_trade self.initial_balance = initial_balance self.normalize_obs = normalize_obs if ensemble_path is None: import config ensemble_path = getattr(config, "ENSEMBLE_MODEL_PATH", "model_ensemble.pkl") self._ensemble = _load_ensemble(ensemble_path) self._obs_dim = OBS_DIM if self._ensemble else (NUM_FEATURES + 5) self.observation_space = spaces.Box( low=-np.inf, high=np.inf, shape=(self._obs_dim,), dtype=np.float32, ) self.action_space = spaces.Discrete(N_ACTIONS) self._all_data: dict = {} self._pairs = pairs or self._detect_pairs() self._load_all_data() # État interne self._candles: list = [] self._features: np.ndarray = np.array([]) self._ml_cache: np.ndarray = np.array([]) # ← CACHE (N, 2) self._step_idx: int = 0 self._ep_end: int = 0 self._pair: str = "" self._in_position: bool = False self._direction: int = 0 self._entry_price: float = 0.0 self._candles_held: int = 0 self._balance: float = initial_balance self._equity_peak: float = initial_balance self._rtracker = RewardTracker() self._rng = np.random.default_rng(seed) log.info(f"[RL-ENV] OBS_DIM={self._obs_dim} | " f"Ensemble={"✅ CACHE" if self._ensemble else "❌ désactivé"}") # ── GYMNASIUM API ───────────────────────────────────────────────────────── def reset(self, seed=None, options=None): super().reset(seed=seed) if seed is not None: self._rng = np.random.default_rng(seed) self._pair = self._rng.choice(self._pairs) data = self._all_data[self._pair] self._candles = data["candles"] self._features = data["features"] self._ml_cache = data["ml_cache"] # ← lookup direct n = len(self._candles) ep_len = int(self._rng.integers(EP_MIN_LEN, min(EP_MAX_LEN, n - WARMUP))) max_start = n - ep_len - WARMUP if max_start <= WARMUP: max_start = WARMUP start = int(self._rng.integers(WARMUP, max(WARMUP + 1, max_start))) self._step_idx = start self._ep_end = start + ep_len self._in_position = False self._direction = 0 self._entry_price = 0.0 self._candles_held = 0 self._balance = self.initial_balance self._equity_peak = self.initial_balance self._rtracker.reset() obs = self._get_obs() info = {"pair": self._pair, "step": self._step_idx} return obs, info def step(self, action: int): assert self.action_space.contains(action), f"Action invalide: {action}" candle = self._candles[self._step_idx] current_price = float(candle["c"]) pnl_realized = 0.0 fee_paid = 0.0 trade_closed = False if self._in_position: pnl_pct = self._calc_pnl_pct(current_price) hit_sl = pnl_pct <= -self.sl_pct hit_tp = pnl_pct >= self.tp_pct timeout = self._candles_held >= MAX_HOLD if hit_sl or hit_tp or timeout: pnl_realized, fee_paid = self._close_position(current_price) trade_closed = True if not trade_closed: if action == 1: if self._in_position and self._direction == -1: pnl_realized, fee_paid = self._close_position(current_price) trade_closed = True if not self._in_position: self._open_position(current_price, direction=1) elif action == 2: if self._in_position and self._direction == 1: pnl_realized, fee_paid = self._close_position(current_price) trade_closed = True if not self._in_position: self._open_position(current_price, direction=-1) elif action == 3: if self._in_position: pnl_realized, fee_paid = self._close_position(current_price) trade_closed = True if self._in_position: self._candles_held += 1 pnl_unrealized = 0.0 if self._in_position: pnl_unrealized = self._calc_pnl_pct(current_price) reward = compute_step_reward( tracker = self._rtracker, pnl_realized = pnl_realized, pnl_unrealized = pnl_unrealized, in_position = self._in_position, trade_closed = trade_closed, fee_paid = fee_paid, ) self._step_idx += 1 terminated = self._step_idx >= self._ep_end truncated = False if self._balance <= self.initial_balance * 0.5: terminated = True reward -= 0.5 obs = self._get_obs() info = { "pair": self._pair, "step": self._step_idx, "balance": self._balance, "in_position": self._in_position, "pnl_realized": pnl_realized, "pnl_unrealized": pnl_unrealized, } if terminated: info["episode_metrics"] = compute_episode_metrics(self._rtracker) return obs, reward, terminated, truncated, info def render(self): candle = self._candles[self._step_idx - 1] print( f"[{self._pair}] Step={self._step_idx} | " f"Price={candle['c']:.5f} | " f"Pos={'LONG' if self._direction==1 else ('SHORT' if self._direction==-1 else 'FLAT')} | " f"Balance={self._balance:.2f}" ) # ── HELPERS PRIVÉS ──────────────────────────────────────────────────────── def _detect_pairs(self) -> list: if not os.path.exists(self.data_dir): raise FileNotFoundError(f"data_dir introuvable : {self.data_dir}") pairs = [ fname.replace("_1h.json", "") for fname in os.listdir(self.data_dir) if fname.endswith("_1h.json") ] if not pairs: raise ValueError(f"Aucun fichier *_1h.json trouvé dans {self.data_dir}") return sorted(pairs) def _load_all_data(self): ref_pair = "EURUSD" ref_path = os.path.join(self.data_dir, f"{ref_pair}_1h.json") ref_closes = None if os.path.exists(ref_path): with open(ref_path) as f: ref_data = json.load(f) ref_closes = np.array([c["c"] for c in ref_data]) loaded = 0 for pair in self._pairs: path = os.path.join(self.data_dir, f"{pair}_1h.json") if not os.path.exists(path): continue with open(path) as f: candles = json.load(f) if len(candles) < MIN_CANDLES: continue btc_c = None if ref_closes is not None and len(ref_closes) >= len(candles): btc_c = ref_closes[-len(candles):] raw_feats = build_features(candles, btc_closes=btc_c) raw_feats = np.nan_to_num( raw_feats, nan=0.0, posinf=0.0, neginf=0.0 ).astype(np.float32) self._all_data[pair] = { "candles": candles, "raw_features": raw_feats, "features": raw_feats.copy(), "ml_cache": None, # rempli après normalisation } loaded += 1 if loaded == 0: raise ValueError("Aucune donnée chargée. Exécuter download_data.py d abord.") self._pairs = sorted(self._all_data.keys()) print(f"[RL-ENV] {loaded} paires chargées — " f"{sum(len(v['candles']) for v in self._all_data.values()):,} candles au total") # Normalisation z-score if self.normalize_obs: all_feats = np.concatenate( [v["features"] for v in self._all_data.values()], axis=0 ) self._feat_mean = all_feats.mean(axis=0) self._feat_std = all_feats.std(axis=0) + 1e-8 for pair in self._all_data: self._all_data[pair]["features"] = ( (self._all_data[pair]["features"] - self._feat_mean) / self._feat_std ).astype(np.float32) print("[RL-ENV] Features normalisées (z-score global)") # ── PRÉ-CALCUL CACHE ML (batch, une seule fois) ────────────────────── if self._ensemble is not None: total = sum(len(v["candles"]) for v in self._all_data.values()) print(f"[RL-ENV] Pré-calcul ML cache ({total:,} candles)...") for pair in self._pairs: raw_f = self._all_data[pair]["raw_features"] cache = _predict_ensemble_batch(self._ensemble, raw_f) self._all_data[pair]["ml_cache"] = cache # (N, 2) float32 print("[RL-ENV] Cache ML prêt — lookup O(1) pendant l entraînement ✅") else: for pair in self._pairs: N = len(self._all_data[pair]["candles"]) self._all_data[pair]["ml_cache"] = np.column_stack([ np.full(N, 0.5, dtype=np.float32), np.zeros(N, dtype=np.float32), ]) def _get_obs(self) -> np.ndarray: idx = self._step_idx feat = self._features[idx].copy() candle_price = float(self._candles[idx]["c"]) upnl = (np.clip(self._calc_pnl_pct(candle_price) / 0.05, -1.0, 1.0) if self._in_position else 0.0) pos_feats = np.array([ float(self._in_position), float(self._direction), upnl, np.clip(self._candles_held / MAX_HOLD, 0.0, 1.0), np.clip(self._balance / self.initial_balance, 0.0, 2.0), ], dtype=np.float32) if self._ensemble is not None: # ← LOOKUP O(1) — zéro appel ML ens_feats = self._ml_cache[idx] # shape (2,) obs = np.concatenate([feat, pos_feats, ens_feats]) else: obs = np.concatenate([feat, pos_feats]) return obs.astype(np.float32) def _open_position(self, price: float, direction: int): self._in_position = True self._direction = direction self._entry_price = price self._candles_held = 0 def _close_position(self, price: float) -> tuple: pnl_pct = self._calc_pnl_pct(price) fee_paid = FEE_RATE * 2 trade_pnl = pnl_pct * self.risk_per_trade self._balance = self._balance * (1.0 + trade_pnl - fee_paid) if self._balance > self._equity_peak: self._equity_peak = self._balance self._in_position = False self._direction = 0 self._entry_price = 0.0 self._candles_held = 0 return pnl_pct, fee_paid def _calc_pnl_pct(self, current_price: float) -> float: if not self._in_position or self._entry_price == 0: return 0.0 raw = (current_price - self._entry_price) / self._entry_price return raw * self._direction * self.leverage # ── PROPRIÉTÉS ──────────────────────────────────────────────────────────── @property def feat_mean(self) -> np.ndarray: return self._feat_mean if self.normalize_obs else np.zeros(NUM_FEATURES) @property def feat_std(self) -> np.ndarray: return self._feat_std if self.normalize_obs else np.ones(NUM_FEATURES) @property def obs_dim(self) -> int: return self._obs_dim