Abdoul Ahad Binizi
364 lines
13 KiB
Python
364 lines
13 KiB
Python
"""
|
||
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 <hours> 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 <window> 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")
|