Abdoul Ahad Binizi
464 lines
18 KiB
Python
464 lines
18 KiB
Python
|
|
"""
|
|
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
|