""" AHAD QUANT — Experience Buffer (Apprentissage Continu V7) ============================================================ Replay buffer thread-safe qui stocke chaque trade fermé avec son contexte complet. Persiste sur disque (experience_buffer.json). Structure d'une expérience : { "id": "20260612_143022_EURUSD", "timestamp": 1718200222.0, "pair": "EURUSD", "signal": "LONG", "confidence": 0.78, "ml_probas": [0.72, 0.75, 0.71, 0.74, 0.73], "rl_action": 1, "rl_agreed": True, "features": [...], # 62 features à l'entrée "regime": "NORMAL", "entry_price": 1.08501, "exit_price": 1.08762, "pnl": 0.0241, "outcome": "WIN", # WIN / LOSS / TIMEOUT "hold_candles": 4, "error_weight": 1.0 # WIN=1.0 TIMEOUT=1.5 LOSS=2.0 } Usage : buffer = ExperienceBuffer() buffer.load() buffer.add(experience) samples = buffer.sample(n=256, error_weighted=True) recent = buffer.get_recent(hours=24) losses = buffer.get_loss_trades(min_weight=1.5) stats = buffer.stats() buffer.save() """ import json import logging import os import random import threading import time from datetime import datetime, timezone from typing import Dict, List, Optional from features import NUM_FEATURES log = logging.getLogger("ExperienceBuffer") # ── Constantes ──────────────────────────────────────────────────────────────── MAX_BUFFER_SIZE = 10_000 # trades max en mémoire (FIFO si dépassé) ERROR_WEIGHT_LOSS = 2.0 # LOSS pèse 2× lors du sampling ERROR_WEIGHT_TIMEOUT = 1.5 # TIMEOUT pèse 1.5× ERROR_WEIGHT_WIN = 1.0 # WIN pèse 1× (normal) BUFFER_FILE = "experience_buffer.json" class ExperienceBuffer: """ Buffer thread-safe de toutes les expériences de trading. Utilisé par OnlineLearner et PerformanceMonitor. """ def __init__( self, max_size: int = MAX_BUFFER_SIZE, buffer_file: str = BUFFER_FILE, ): self.max_size = max_size self.buffer_file = buffer_file self._experiences: List[Dict] = [] self._lock = threading.RLock() # ── I/O ────────────────────────────────────────────────────────────────── def save(self) -> bool: """Sauvegarde le buffer sur disque. Retourne True si succès.""" with self._lock: try: tmp = self.buffer_file + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(self._experiences, f, ensure_ascii=False) os.replace(tmp, self.buffer_file) log.debug(f"Buffer sauvegardé ({len(self._experiences)} trades)") return True except Exception as e: log.error(f"Erreur sauvegarde buffer : {e}") return False def load(self) -> bool: """ Charge le buffer depuis le disque. Fail-safe : si le fichier est corrompu → buffer vide, trading continue. """ if not os.path.exists(self.buffer_file): log.info("Aucun buffer existant — démarrage avec buffer vide") return False try: with open(self.buffer_file, encoding="utf-8") as f: data = json.load(f) if not isinstance(data, list): raise ValueError("Format invalide — doit être une liste") with self._lock: self._experiences = data[-self.max_size:] log.info(f"Buffer chargé : {len(self._experiences)} trades") return True except Exception as e: log.error(f"Erreur chargement buffer (reset) : {e}") with self._lock: self._experiences = [] return False # ── Ajout ──────────────────────────────────────────────────────────────── def add(self, experience: Dict) -> None: """ Ajoute une expérience au buffer. - Complète les champs manquants (id, timestamp, error_weight). - Tronque le buffer si MAX_BUFFER_SIZE est dépassé (FIFO). - Sauvegarde automatiquement toutes les 100 trades. """ with self._lock: # Champs obligatoires avec valeurs par défaut exp = dict(experience) if "id" not in exp: ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") exp["id"] = f"{ts}_{exp.get('pair','UNKNOWN')}" if "timestamp" not in exp: exp["timestamp"] = time.time() if "error_weight" not in exp: outcome = exp.get("outcome", "WIN") exp["error_weight"] = _outcome_to_weight(outcome) self._experiences.append(exp) # FIFO : supprimer les plus anciens si dépassement if len(self._experiences) > self.max_size: excess = len(self._experiences) - self.max_size self._experiences = self._experiences[excess:] # Auto-sauvegarde toutes les 100 trades if len(self._experiences) % 100 == 0: self.save() # ── Sampling ───────────────────────────────────────────────────────────── def sample(self, n: int = 256, error_weighted: bool = True) -> List[Dict]: """ Retourne n expériences aléatoires. Si error_weighted=True : les LOSS et TIMEOUT ont plus de chances d'être sélectionnés. """ with self._lock: pool = list(self._experiences) if not pool: return [] n = min(n, len(pool)) if not error_weighted: return random.sample(pool, n) # Pondération par error_weight weights = [e.get("error_weight", 1.0) for e in pool] total = sum(weights) probs = [w / total for w in weights] indices = random.choices(range(len(pool)), weights=probs, k=n) return [pool[i] for i in indices] # ── Filtres ────────────────────────────────────────────────────────────── def get_recent( self, hours: float = 24.0, outcome_filter: Optional[List[str]] = None, ) -> List[Dict]: """ Retourne les trades des dernières heures. Si outcome_filter est fourni (ex: ["LOSS","TIMEOUT"]), filtre par outcome. """ cutoff = time.time() - hours * 3600 with self._lock: pool = [e for e in self._experiences if e.get("timestamp", 0) >= cutoff] if outcome_filter: pool = [e for e in pool if e.get("outcome", "") in outcome_filter] return pool def get_loss_trades(self, min_weight: float = 1.5) -> List[Dict]: """Retourne les trades avec error_weight >= min_weight (LOSS + TIMEOUT).""" with self._lock: return [e for e in self._experiences if e.get("error_weight", 1.0) >= min_weight] def get_by_outcome(self, outcome: str) -> List[Dict]: """Retourne tous les trades d'un outcome donné (WIN/LOSS/TIMEOUT).""" with self._lock: return [e for e in self._experiences if e.get("outcome") == outcome] def get_all(self) -> List[Dict]: """Retourne une copie de tout le buffer.""" with self._lock: return list(self._experiences) def get_last_n(self, n: int) -> List[Dict]: """Retourne les n derniers trades.""" with self._lock: return list(self._experiences[-n:]) # ── Statistiques ───────────────────────────────────────────────────────── def stats(self, window: int = 0) -> Dict: """ Retourne les statistiques du buffer. Si window > 0, calcule sur les derniers trades. """ with self._lock: data = list(self._experiences[-window:]) if window > 0 else list(self._experiences) if not data: return { "total_trades": 0, "win_rate": 0.0, "loss_rate": 0.0, "timeout_rate": 0.0, "avg_pnl": 0.0, "total_pnl": 0.0, "max_drawdown": 0.0, "avg_confidence": 0.0, "avg_hold_candles": 0.0, } total = len(data) wins = sum(1 for e in data if e.get("outcome") == "WIN") losses = sum(1 for e in data if e.get("outcome") == "LOSS") timeouts = sum(1 for e in data if e.get("outcome") == "TIMEOUT") pnls = [e.get("pnl", 0.0) for e in data] confs = [e.get("confidence", 0.0) for e in data if e.get("confidence")] holds = [e.get("hold_candles", 0) for e in data] # Max drawdown (peak-to-trough sur la séquence de PnL cumulatif) cumulative = 0.0 peak = 0.0 max_dd = 0.0 for p in pnls: cumulative += p if cumulative > peak: peak = cumulative dd = (peak - cumulative) / (abs(peak) + 1e-9) if dd > max_dd: max_dd = dd return { "total_trades": total, "win_rate": wins / total, "loss_rate": losses / total, "timeout_rate": timeouts / total, "avg_pnl": sum(pnls) / total, "total_pnl": sum(pnls), "max_drawdown": max_dd, "avg_confidence": sum(confs) / len(confs) if confs else 0.0, "avg_hold_candles": sum(holds) / len(holds) if holds else 0.0, } @property def total_trades(self) -> int: with self._lock: return len(self._experiences) def __len__(self) -> int: return self.total_trades def __repr__(self) -> str: s = self.stats() return ( f"ExperienceBuffer(" f"trades={s['total_trades']}, " f"win_rate={s['win_rate']:.1%}, " f"avg_pnl={s['avg_pnl']:.4f})" ) # ── Utilitaire ──────────────────────────────────────────────────────────────── def _outcome_to_weight(outcome: str) -> float: """Convertit un outcome en error_weight.""" return { "WIN": ERROR_WEIGHT_WIN, "LOSS": ERROR_WEIGHT_LOSS, "TIMEOUT": ERROR_WEIGHT_TIMEOUT, }.get(outcome.upper(), 1.0) # ── Singleton global (utilisé par ahad_quant.py, online_learner.py, etc.) ───── _global_buffer: Optional[ExperienceBuffer] = None _buffer_lock = threading.Lock() def get_experience_buffer( max_size: int = MAX_BUFFER_SIZE, buffer_file: str = BUFFER_FILE, auto_load: bool = True, ) -> ExperienceBuffer: """ Retourne le singleton global du buffer. Thread-safe. Charge automatiquement depuis le disque au premier appel. """ global _global_buffer with _buffer_lock: if _global_buffer is None: _global_buffer = ExperienceBuffer(max_size=max_size, buffer_file=buffer_file) if auto_load: _global_buffer.load() return _global_buffer # ── CLI de test ─────────────────────────────────────────────────────────────── if __name__ == "__main__": import logging logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(name)s] %(message)s") buf = ExperienceBuffer(max_size=100, buffer_file="test_buffer.json") buf.load() print(f"Buffer chargé : {buf.total_trades} trades") # Ajouter des trades de test for i in range(20): outcome = ["WIN", "LOSS", "TIMEOUT"][i % 3] buf.add({ "pair": "EURUSD", "signal": "LONG" if i % 2 == 0 else "SHORT", "confidence": round(0.72 + (i % 5) * 0.02, 3), "ml_probas": [0.72, 0.73, 0.71, 0.74, 0.72], "rl_action": 1, "rl_agreed": True, "features": [0.0] * NUM_FEATURES, "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, }) print(f"\n{buf}") print(f"\nStats : {json.dumps(buf.stats(), indent=2)}") print(f"\n5 derniers trades : {[e['outcome'] for e in buf.get_last_n(5)]}") print(f"LOSS trades : {len(buf.get_loss_trades())} (weight >= 1.5)") print(f"Récents 24h : {len(buf.get_recent(hours=24))} trades") # Nettoyage test if os.path.exists("test_buffer.json"): os.remove("test_buffer.json") print("\n✅ ExperienceBuffer — test OK")