Abdoul Ahad Binizi
804 lines
33 KiB
Python
804 lines
33 KiB
Python
"""
|
||
AHAD QUANT — Auto-Retraining Scheduler
|
||
Re-downloads data and retrains the ensemble model on a schedule.
|
||
|
||
Activated via .env:
|
||
AUTO_RETRAIN_ENABLED=true
|
||
AUTO_RETRAIN_INTERVAL_HOURS=24
|
||
|
||
Logic: the new model replaces the old one ONLY if it is strictly
|
||
more accurate than the currently active model. The active model's
|
||
accuracy is always saved in last_retrain.json and used as the
|
||
comparison baseline — not a fixed threshold.
|
||
"""
|
||
|
||
import os, time, json, pickle, shutil, logging, threading, subprocess, sys
|
||
from datetime import datetime, timezone
|
||
import config
|
||
|
||
log = logging.getLogger("AutoRetrain")
|
||
|
||
# [FIX] Lock global : empêche daily-local et full-retrain de s'exécuter
|
||
# simultanément et d'écrire model_ensemble.pkl en même temps.
|
||
# Toute fonction qui modifie un fichier modèle DOIT acquérir ce lock.
|
||
_retrain_lock = threading.Lock()
|
||
TIMESTAMP_FILE = "last_retrain.json"
|
||
# [FIX 21/06/2026] Constante manquante — provoquait un NameError dans
|
||
# _load_daily_local_state()/_save_daily_local_state() dès le premier appel
|
||
# (should_run_daily_local_cycle() l'appelle sans condition au tout premier
|
||
# tick de AutoRetrainer._loop()), tuant silencieusement le thread de fond
|
||
# entier : ni le cycle quotidien local (warm-start ML + RL real-only), ni
|
||
# le full-retrain réseau ne s'exécutaient jamais en pratique.
|
||
DAILY_LOCAL_TIMESTAMP_FILE = "last_daily_local.json"
|
||
|
||
|
||
def _safe_export_unified():
|
||
"""
|
||
Régénère ahad_quant_unified.zip (best-effort, non-bloquant) après un
|
||
ré-entraînement réussi. N'est PLUS critique pour la correction en live
|
||
depuis le correctif du point #2 du diagnostic (rl_agent.py priorise déjà
|
||
les fichiers vivants rl_agent.zip/rl_scaler.pkl/model_ensemble.pkl, et
|
||
unified_brain.py les recharge à chaud) — reste utile pour des
|
||
déploiements portables sur une autre machine.
|
||
|
||
export_unified.export_unified() appelle sys.exit(1) en interne si les
|
||
fichiers source sont absents : on attrape donc BaseException (et pas
|
||
seulement Exception) pour ne jamais faire planter le cycle de retrain
|
||
à cause de cet export secondaire.
|
||
"""
|
||
try:
|
||
import export_unified
|
||
export_unified.export_unified()
|
||
log.info("[EXPORT] ahad_quant_unified.zip régénéré")
|
||
except BaseException as e:
|
||
log.warning(f"[EXPORT] Échec régénération ahad_quant_unified.zip (non-bloquant) : {e}")
|
||
|
||
|
||
# ── Persistence helpers ───────────────────────────────────────────────────────
|
||
|
||
def _load_state() -> dict:
|
||
"""
|
||
Load the saved state of the currently active model.
|
||
Returns dict with keys: timestamp, accuracy, datetime.
|
||
Returns defaults (accuracy=0) if no state file exists yet.
|
||
"""
|
||
if not os.path.exists(TIMESTAMP_FILE):
|
||
return {"timestamp": 0.0, "accuracy": 0.0, "datetime": "never"}
|
||
try:
|
||
with open(TIMESTAMP_FILE) as f:
|
||
return json.load(f)
|
||
except Exception:
|
||
return {"timestamp": 0.0, "accuracy": 0.0, "datetime": "never"}
|
||
|
||
|
||
def _save_state(accuracy: float):
|
||
"""Save the accuracy of the newly active model."""
|
||
with open(TIMESTAMP_FILE, "w") as f:
|
||
json.dump({
|
||
"timestamp": time.time(),
|
||
"accuracy": round(accuracy, 6),
|
||
"datetime": datetime.now(timezone.utc).isoformat(),
|
||
}, f, indent=2)
|
||
|
||
|
||
def get_active_accuracy() -> float:
|
||
"""Return the accuracy of the currently active model (0 if unknown)."""
|
||
return _load_state().get("accuracy", 0.0)
|
||
|
||
|
||
def should_retrain() -> bool:
|
||
"""True if enough time has passed since last retrain."""
|
||
if not config.AUTO_RETRAIN_ENABLED:
|
||
return False
|
||
elapsed = time.time() - _load_state().get("timestamp", 0.0)
|
||
return elapsed >= config.AUTO_RETRAIN_INTERVAL_HOURS * 3600
|
||
|
||
|
||
# ── Parse accuracy from train.py output ──────────────────────────────────────
|
||
|
||
def _parse_accuracy(output: str) -> float:
|
||
"""Extract ensemble (or LightGBM) accuracy from train.py stdout."""
|
||
accuracy = 0.0
|
||
for line in output.splitlines():
|
||
# Ensemble line has priority
|
||
if "Ensemble test accuracy:" in line:
|
||
try:
|
||
val = line.split(":")[-1].strip().split()[0]
|
||
accuracy = float(val.replace("%", "")) / (100 if "%" in val else 1)
|
||
return accuracy
|
||
except Exception:
|
||
pass
|
||
# Fallback: single model line
|
||
if "Test accuracy:" in line:
|
||
try:
|
||
val = line.split(":")[-1].strip().split()[0]
|
||
accuracy = float(val.replace("%", "")) / (100 if "%" in val else 1)
|
||
except Exception:
|
||
pass
|
||
return accuracy
|
||
|
||
|
||
# ── Main retrain cycle ────────────────────────────────────────────────────────
|
||
|
||
def run_retrain(notify_fn=None) -> dict:
|
||
"""
|
||
Execute a full retrain cycle.
|
||
[FIX] Protégé par _retrain_lock — ne peut pas tourner en même temps
|
||
qu'un cycle daily-local.
|
||
|
||
Decision logic:
|
||
new_accuracy > active_accuracy → replace model ✅
|
||
new_accuracy ≤ active_accuracy → keep old model ❌
|
||
|
||
Returns dict: {success, new_accuracy, old_accuracy, improved, message}
|
||
"""
|
||
if not _retrain_lock.acquire(blocking=False):
|
||
log.warning("[FULL-RETRAIN] Cycle daily-local en cours — full-retrain reporté")
|
||
return {"success": False, "message": "retrain_lock — réessayer dans quelques minutes"}
|
||
try:
|
||
return _run_retrain_inner(notify_fn)
|
||
finally:
|
||
_retrain_lock.release()
|
||
|
||
|
||
def _run_retrain_inner(notify_fn=None) -> dict:
|
||
"""Corps du full retrain (appelé uniquement si lock acquis)."""
|
||
result = {
|
||
"success": False,
|
||
"new_accuracy": 0.0,
|
||
"old_accuracy": 0.0,
|
||
"improved": False,
|
||
"message": "",
|
||
}
|
||
|
||
# Accuracy of the model currently in production
|
||
old_accuracy = get_active_accuracy()
|
||
result["old_accuracy"] = old_accuracy
|
||
|
||
old_label = f"{old_accuracy:.2%}" if old_accuracy > 0 else "unknown (first run)"
|
||
log.info(f"Auto-retrain starting — active model accuracy: {old_label}")
|
||
|
||
if notify_fn:
|
||
notify_fn(
|
||
f"🔄 *Auto-retrain started*\n"
|
||
f"Active model accuracy: `{old_label}`\n"
|
||
f"Downloading fresh data..."
|
||
)
|
||
|
||
# ── Step 1: Download data ────────────────────────────────────────────────
|
||
# Timeout généreux : 27 coins × 1000 jours peut prendre 30–60 min selon
|
||
# la connexion et les rate-limits de l'exchange (14400s = 4h max).
|
||
try:
|
||
proc = subprocess.run(
|
||
[sys.executable, "download_data.py"],
|
||
capture_output=True, text=True, timeout=14400
|
||
)
|
||
if proc.returncode != 0:
|
||
raise RuntimeError(proc.stderr[-500:] or "download failed")
|
||
log.info("Data download complete")
|
||
except subprocess.TimeoutExpired:
|
||
msg = "❌ Auto-retrain: download timed out after 4h — vérifie la connexion ou réduis la liste de coins"
|
||
log.error(msg)
|
||
if notify_fn: notify_fn(msg)
|
||
result["message"] = msg
|
||
return result
|
||
except Exception as e:
|
||
msg = f"❌ Auto-retrain: data download failed — {e}"
|
||
log.error(msg)
|
||
if notify_fn: notify_fn(msg)
|
||
result["message"] = msg
|
||
return result
|
||
|
||
# ── Step 2: Train new model ──────────────────────────────────────────────
|
||
try:
|
||
proc = subprocess.run(
|
||
[sys.executable, "train.py"],
|
||
capture_output=True, text=True, timeout=7200
|
||
)
|
||
output = proc.stdout + proc.stderr
|
||
new_accuracy = _parse_accuracy(output)
|
||
result["new_accuracy"] = new_accuracy
|
||
|
||
if proc.returncode != 0:
|
||
raise RuntimeError(output[-500:])
|
||
|
||
log.info(f"Training complete — new accuracy: {new_accuracy:.2%}")
|
||
except Exception as e:
|
||
msg = f"❌ Auto-retrain: training failed — {e}"
|
||
log.error(msg)
|
||
if notify_fn: notify_fn(msg)
|
||
result["message"] = msg
|
||
return result
|
||
|
||
# ── Step 3: Compare strictly against active model ────────────────────────
|
||
improved = new_accuracy > old_accuracy
|
||
result["improved"] = improved
|
||
|
||
if improved:
|
||
# New model is better → save it as active
|
||
_save_state(new_accuracy)
|
||
result["success"] = True
|
||
|
||
gain = new_accuracy - old_accuracy
|
||
msg = (
|
||
f"✅ *Model updated*\n"
|
||
f"Old accuracy: `{old_label}`\n"
|
||
f"New accuracy: `{new_accuracy:.2%}` (+{gain:.2%})\n"
|
||
f"New model is now active."
|
||
)
|
||
log.info(msg.replace("*","").replace("`",""))
|
||
if notify_fn: notify_fn(msg)
|
||
result["message"] = msg
|
||
|
||
else:
|
||
# New model is not better → keep old model (train.py already wrote
|
||
# the new .pkl files; we restore the backup if it exists, otherwise
|
||
# we simply do nothing since train.py didn't overwrite if accuracy
|
||
# was logged as lower — but to be safe we log a warning)
|
||
gap = old_accuracy - new_accuracy
|
||
msg = (
|
||
f"⚠️ *Model NOT updated*\n"
|
||
f"Old accuracy: `{old_label}`\n"
|
||
f"New accuracy: `{new_accuracy:.2%}` (-{gap:.2%})\n"
|
||
f"Active model unchanged — old model was better."
|
||
)
|
||
log.warning(msg.replace("*","").replace("`",""))
|
||
if notify_fn: notify_fn(msg)
|
||
result["message"] = msg
|
||
|
||
# Restore backup if it was created before training
|
||
for path in [config.MODEL_PATH, config.ENSEMBLE_MODEL_PATH]:
|
||
backup = path + ".backup"
|
||
if os.path.exists(backup):
|
||
shutil.copy2(backup, path)
|
||
log.info(f"Restored backup: {path}")
|
||
|
||
# ── RL — TOUJOURS exécuté après le ML, que le nouveau ML ait été accepté
|
||
# ou non (le RL doit de toute façon rester synchronisé avec l'ensemble
|
||
# actif, ancien ou nouveau). C'est ce qui garantit que ce chemin
|
||
# "full retrain réseau" ne redevient JAMAIS ML-seul : ML et RL forment
|
||
# un seul cycle, même ici. Le fine-tune RL a son propre accept/reject
|
||
# (voir rl_train.py::fine_tune()), donc aucun risque de régression. ──
|
||
log.info("[FULL-RETRAIN] Fine-tuning RL sur l'ensemble ML actif...")
|
||
try:
|
||
rl_updated = run_rl_retrain_with_replay(notify_fn)
|
||
except Exception as e:
|
||
log.warning(f"[FULL-RETRAIN] Fine-tuning RL échoué (non-bloquant) : {e}")
|
||
rl_updated = False
|
||
result["rl_updated"] = rl_updated
|
||
|
||
# ── Export unifié — UNE SEULE fois, après ML ET RL, pour ne jamais
|
||
# bundler un ML neuf avec un RL périmé (ou l'inverse). ──
|
||
_safe_export_unified()
|
||
|
||
return result
|
||
|
||
|
||
# ── Backup helper (called before training to preserve old model) ───────────────
|
||
|
||
def _backup_models():
|
||
"""Create .backup copies of model files before overwriting."""
|
||
for path in [config.MODEL_PATH, config.ENSEMBLE_MODEL_PATH]:
|
||
if os.path.exists(path):
|
||
shutil.copy2(path, path + ".backup")
|
||
log.debug(f"Backed up: {path}")
|
||
|
||
|
||
# ── Background scheduler ──────────────────────────────────────────────────────
|
||
|
||
class AutoRetrainer:
|
||
"""
|
||
Background thread that periodically retrains the model.
|
||
Integrates with the running AHAD QUANT bot.
|
||
"""
|
||
|
||
def __init__(self, notify_fn=None):
|
||
self.notify_fn = notify_fn
|
||
self._thread = None
|
||
self._stop_event = threading.Event()
|
||
|
||
def start(self):
|
||
if not config.AUTO_RETRAIN_ENABLED and not getattr(config, "DAILY_LOCAL_RETRAIN_ENABLED", True):
|
||
log.info("Auto-retrain entièrement désactivé (AUTO_RETRAIN_ENABLED=false et DAILY_LOCAL_RETRAIN_ENABLED=false)")
|
||
return
|
||
self._thread = threading.Thread(
|
||
target=self._loop, daemon=True, name="AutoRetrain"
|
||
)
|
||
self._thread.start()
|
||
active = get_active_accuracy()
|
||
label = f"{active:.2%}" if active > 0 else "unknown (first run)"
|
||
daily_interval = getattr(config, "DAILY_LOCAL_RETRAIN_INTERVAL_HOURS", 24)
|
||
log.info(
|
||
f"Auto-retrain scheduler started "
|
||
f"(cycle quotidien local ML+RL toutes les {daily_interval}h"
|
||
+ (f" | ré-entraînement complet réseau toutes les {config.AUTO_RETRAIN_INTERVAL_HOURS}h"
|
||
if config.AUTO_RETRAIN_ENABLED else " | ré-entraînement complet réseau désactivé")
|
||
+ f" | active model: {label})"
|
||
)
|
||
|
||
def stop(self):
|
||
self._stop_event.set()
|
||
|
||
def _loop(self):
|
||
while not self._stop_event.is_set():
|
||
# ── Cycle quotidien LOCAL (ML + RL unifiés) — c'est le chemin
|
||
# automatique principal demandé : 100% local, journalier, et
|
||
# n'accepte un nouveau modèle (ML ou RL) que s'il est meilleur.
|
||
if should_run_daily_local_cycle():
|
||
log.info("[DAILY-LOCAL] Intervalle atteint — démarrage du cycle quotidien local")
|
||
run_daily_unified_retrain(self.notify_fn)
|
||
|
||
# ── Ré-entraînement complet (lourd, réseau) — optionnel, désactivé
|
||
# par défaut (AUTO_RETRAIN_ENABLED=false) car il télécharge de
|
||
# nouvelles données ; reste disponible si explicitement activé.
|
||
if config.AUTO_RETRAIN_ENABLED and should_retrain():
|
||
log.info("[FULL-RETRAIN] Intervalle atteint — backup et ré-entraînement complet (réseau)")
|
||
_backup_models()
|
||
run_retrain(self.notify_fn)
|
||
|
||
# Check every 30 minutes
|
||
self._stop_event.wait(timeout=1800)
|
||
|
||
|
||
# ── CLI ───────────────────────────────────────────────────────────────────────
|
||
|
||
if __name__ == "__main__":
|
||
logging.basicConfig(level=logging.INFO,
|
||
format="%(asctime)s [%(name)s] %(message)s")
|
||
|
||
def notify(msg):
|
||
print(f"\n[NOTIFY] {msg}\n")
|
||
|
||
state = _load_state()
|
||
print("=" * 55)
|
||
print(" AHAD QUANT — Auto-Retrain")
|
||
print("=" * 55)
|
||
print(f" Active model accuracy : "
|
||
f"{state['accuracy']:.2%}" if state['accuracy'] else " Active model: unknown (first run)")
|
||
print(f" Last retrain : {state.get('datetime','never')}")
|
||
print(f" Interval : every {config.AUTO_RETRAIN_INTERVAL_HOURS}h")
|
||
print(f" Rule : new model must beat active model")
|
||
print("=" * 55 + "\n")
|
||
|
||
print("Backing up current models...")
|
||
_backup_models()
|
||
|
||
print("Running retrain now...\n")
|
||
result = run_retrain(notify)
|
||
|
||
print("\n" + "=" * 55)
|
||
print(f" Old accuracy : {result['old_accuracy']:.2%}")
|
||
print(f" New accuracy : {result['new_accuracy']:.2%}")
|
||
print(f" Improved : {'✅ YES — model updated' if result['improved'] else '❌ NO — old model kept'}")
|
||
print("=" * 55)
|
||
|
||
|
||
# ─── RL Agent Auto-Retrain ────────────────────────────────────────────────────
|
||
|
||
RL_TIMESTAMP_FILE = "last_rl_retrain.json"
|
||
|
||
|
||
def _load_rl_state() -> dict:
|
||
if not os.path.exists(RL_TIMESTAMP_FILE):
|
||
return {"timestamp": 0.0, "datetime": "never"}
|
||
try:
|
||
with open(RL_TIMESTAMP_FILE) as f:
|
||
return json.load(f)
|
||
except Exception:
|
||
return {"timestamp": 0.0, "datetime": "never"}
|
||
|
||
|
||
def _save_rl_state():
|
||
with open(RL_TIMESTAMP_FILE, "w") as f:
|
||
json.dump({
|
||
"timestamp": time.time(),
|
||
"datetime": datetime.now(timezone.utc).isoformat(),
|
||
}, f, indent=2)
|
||
|
||
|
||
def should_retrain_rl() -> bool:
|
||
"""True si le fine-tuning RL hebdomadaire est dû."""
|
||
if not getattr(config, "RL_AUTO_RETRAIN_ENABLED", False):
|
||
return False
|
||
interval = getattr(config, "RL_RETRAIN_INTERVAL_HOURS", 168) * 3600
|
||
elapsed = time.time() - _load_rl_state().get("timestamp", 0.0)
|
||
return elapsed >= interval
|
||
|
||
|
||
def run_rl_finetune() -> dict:
|
||
"""
|
||
Lance le fine-tuning RL en sous-processus (rl_train.py --finetune).
|
||
|
||
Depuis le correctif rl_train.py::fine_tune(), le sous-processus évalue
|
||
l'ancienne et la nouvelle politique sur les mêmes épisodes (seed fixe)
|
||
et n'écrase rl_agent.zip QUE si la nouvelle est meilleure — exactement
|
||
la même philosophie "si meilleur on remplace, sinon on conserve" déjà
|
||
appliquée côté ML. Un fine-tune REJETÉ (ancienne politique conservée)
|
||
est un résultat NORMAL, pas un échec — distinct d'un crash du
|
||
sous-processus (returncode != 0).
|
||
|
||
Retourne {"success": bool, "accepted": bool, "old_mean_reward": float|None,
|
||
"new_mean_reward": float|None}.
|
||
"""
|
||
model_path = getattr(config, "RL_MODEL_PATH", "rl_agent")
|
||
steps = getattr(config, "RL_FINETUNE_STEPS", 200_000)
|
||
|
||
result = {"success": False, "accepted": False, "old_mean_reward": None, "new_mean_reward": None}
|
||
|
||
if not os.path.exists(f"{model_path}.zip"):
|
||
log.warning("[RL-RETRAIN] Modèle rl_agent.zip introuvable — fine-tune ignoré")
|
||
return result
|
||
|
||
log.info(f"[RL-RETRAIN] Démarrage fine-tune RL (+{steps:,} steps)...")
|
||
try:
|
||
proc = subprocess.run(
|
||
[sys.executable, "rl_train.py",
|
||
"--finetune",
|
||
"--model", model_path,
|
||
"--steps", str(steps)],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=7200, # 2h max
|
||
)
|
||
if proc.returncode != 0:
|
||
log.error(f"[RL-RETRAIN] Échec fine-tune RL:\n{proc.stderr[-500:]}")
|
||
return result
|
||
|
||
result["success"] = True
|
||
|
||
# Parse le marqueur "RL_FINETUNE_RESULT accepted=... old=... new=..."
|
||
# émis par rl_train.py::fine_tune() — seule façon pour ce process
|
||
# parent de connaître le résultat accept/reject du sous-processus.
|
||
accepted = False
|
||
for line in proc.stdout.splitlines():
|
||
if line.startswith("RL_FINETUNE_RESULT"):
|
||
try:
|
||
parts = dict(p.split("=") for p in line.split()[1:])
|
||
accepted = parts.get("accepted", "False") == "True"
|
||
result["old_mean_reward"] = float(parts.get("old", "nan"))
|
||
result["new_mean_reward"] = float(parts.get("new", "nan"))
|
||
except Exception:
|
||
pass
|
||
break
|
||
result["accepted"] = accepted
|
||
|
||
if accepted:
|
||
log.info(
|
||
f"[RL-RETRAIN] ✅ Fine-tune accepté "
|
||
f"(reward {result['old_mean_reward']:.4f} → {result['new_mean_reward']:.4f})"
|
||
)
|
||
_save_rl_state()
|
||
# Recharger le singleton rl_agent en mémoire — le fichier a changé
|
||
try:
|
||
from rl_agent import reset_rl_agent
|
||
reset_rl_agent()
|
||
log.info("[RL-RETRAIN] Singleton RL rechargé")
|
||
except ImportError:
|
||
pass
|
||
_safe_export_unified()
|
||
else:
|
||
log.info(
|
||
f"[RL-RETRAIN] ↔️ Fine-tune rejeté — politique actuelle conservée "
|
||
f"(reward {result['old_mean_reward']:.4f} vs {result['new_mean_reward']:.4f} proposé)"
|
||
)
|
||
|
||
return result
|
||
except subprocess.TimeoutExpired:
|
||
log.error("[RL-RETRAIN] Timeout fine-tune RL (>2h)")
|
||
return result
|
||
except Exception as e:
|
||
log.error(f"[RL-RETRAIN] Exception : {e}")
|
||
return result
|
||
|
||
|
||
# ─── Apprentissage Continu V7 — Cycle quotidien LGB + Replay RL ──────────────
|
||
|
||
def run_daily_lightgbm_warmstart(notify_fn=None) -> bool:
|
||
"""
|
||
Cycle quotidien léger (local CPU, ~2-5 min).
|
||
Warm-start LightGBM sur les trades LOSS + TIMEOUT.
|
||
|
||
[FIX] 3 améliorations :
|
||
1. Fenêtre progressive : si pas assez de trades en 24h, on élargit à 48h,
|
||
72h, puis tout le buffer — les phases de bonne performance ne bloquent
|
||
plus le cycle.
|
||
2. Tous les types de trades sont inclus (WIN surpondéré 1x, LOSS 2x,
|
||
TIMEOUT 1.5x via error_weight déjà dans le buffer) — plus de dépendance
|
||
à un minimum de pertes.
|
||
3. Modèle candidat accumulatif : si le warmstart est rejeté, les arbres
|
||
appris sont sauvegardés dans model.pkl.candidate pour être utilisés comme
|
||
point de départ du prochain cycle (au lieu de repartir de la production).
|
||
L'apprentissage des mauvais trades s'accumule même quand le modèle actuel
|
||
reste en place.
|
||
"""
|
||
if not getattr(config, "CONTINUOUS_LEARNING_ENABLED", False):
|
||
return False
|
||
if not getattr(config, "WARMSTART_ENABLED", False):
|
||
return False
|
||
|
||
try:
|
||
from experience_buffer import get_experience_buffer
|
||
from online_learner import OnlineLearner
|
||
|
||
buf = get_experience_buffer(auto_load=True)
|
||
learner = OnlineLearner(buf)
|
||
|
||
min_trades = getattr(config, "WARMSTART_MIN_TRADES", 5)
|
||
|
||
# ── Fenêtre progressive : 24h → 48h → 72h → tout le buffer ──────────
|
||
trades = []
|
||
window_hours = 24
|
||
for window_hours in [24, 48, 72, 9999]:
|
||
if window_hours == 9999:
|
||
trades = buf.get_recent(hours=87600) # ~10 ans = tout le buffer
|
||
else:
|
||
trades = buf.get_recent(hours=window_hours)
|
||
if len(trades) >= min_trades:
|
||
break
|
||
|
||
n = len(trades)
|
||
if n < min_trades:
|
||
log.info(f"[CL-DAILY] Buffer total seulement {n} trades (min={min_trades}) — skip warm-start")
|
||
return False
|
||
|
||
window_label = f"{window_hours}h" if window_hours != 9999 else "tout le buffer"
|
||
log.info(f"[CL-DAILY] Warm-start LGB sur {n} trades ({window_label})...")
|
||
updated = learner.daily_update(loss_trades=trades)
|
||
|
||
msg = (
|
||
f"{'✅' if updated else '⚠️'} *CL Warm-start quotidien*\n"
|
||
f"Trades utilisés : {n} ({window_label})\n"
|
||
f"Modèle {'mis à jour' if updated else 'inchangé (candidate sauvegardé)'}"
|
||
)
|
||
log.info(msg.replace("*","").replace("`",""))
|
||
if notify_fn:
|
||
notify_fn(msg)
|
||
|
||
if updated:
|
||
_safe_export_unified()
|
||
|
||
return updated
|
||
|
||
except Exception as e:
|
||
log.error(f"[CL-DAILY] Erreur warm-start : {e}")
|
||
return False
|
||
|
||
|
||
def run_rl_retrain_with_replay(notify_fn=None) -> bool:
|
||
"""
|
||
Cycle RL quotidien — 100% basé sur les vrais trades du bot, ZERO
|
||
simulation dans l'entrainement.
|
||
|
||
[CORRECTIF 21/06/2026] AVANT : cette fonction lancait TOUJOURS
|
||
run_rl_finetune() en etape 1 -- un fine-tune PPO classique sur
|
||
l'environnement SIMULE (AhadQuantForexEnv, donnees de marche
|
||
historiques) -- puis tentait d'injecter les vrais trades en etape 2.
|
||
Cette injection ne servait a rien : rl_train.py::fine_tune() appelait
|
||
model.learn(), qui reinitialise le rollout_buffer
|
||
(rollout_buffer.reset() en tete de collect_rollouts()) AVANT tout
|
||
entrainement. Le RL n'apprenait donc QUE de la simulation, jamais des
|
||
vrais trades -- meme symptome que le bug RL/backtest deja corrige.
|
||
|
||
MAINTENANT : si le buffer contient >= RL_REPLAY_MIN_TRADES (config,
|
||
defaut 50) trades reels, on lance rl_train.py --finetune --replay
|
||
<fichier>, qui appelle fine_tune_real_only() : un model.train() PPO
|
||
direct sur un rollout_buffer construit a 100% a partir des vrais
|
||
trades (voir rl_train.py). Aucun appel a run_rl_finetune() / a
|
||
AhadQuantForexEnv dans ce cycle.
|
||
|
||
Si le nombre de trades reels est insuffisant : le cycle RL est
|
||
SKIP cette fois-ci -- PAS de repli sur la simulation. Le RL ne doit
|
||
progresser que sur ce que le bot a reellement fait ; tant qu'il n'y a
|
||
pas assez de volume reel, on attend plutot que d'entrainer sur du
|
||
synthetique.
|
||
|
||
Retourne True si le fine-tune reel a ete accepte (modele mis a jour).
|
||
"""
|
||
if not getattr(config, "CONTINUOUS_LEARNING_ENABLED", False):
|
||
return False
|
||
if not getattr(config, "RL_REAL_REPLAY_ENABLED", False):
|
||
return False
|
||
|
||
try:
|
||
from experience_buffer import get_experience_buffer
|
||
from online_learner import OnlineLearner
|
||
|
||
buf = get_experience_buffer(auto_load=True)
|
||
learner = OnlineLearner(buf)
|
||
|
||
min_trades = getattr(config, "RL_REPLAY_MIN_TRADES", 50)
|
||
if buf.total_trades < min_trades:
|
||
log.info(f"[CL-RL] Seulement {buf.total_trades} trades reels dans le buffer "
|
||
f"(min={min_trades}) -- RL update SKIP ce cycle (aucun repli simule)")
|
||
return False
|
||
|
||
episodes = learner.prepare_rl_episodes(min_trades=min_trades)
|
||
if not episodes:
|
||
log.info("[CL-RL] Aucun episode reel valide prepare -- RL update SKIP")
|
||
return False
|
||
|
||
# Sauvegarder les episodes pour injection dans rl_train.py
|
||
import json, os
|
||
replay_file = "rl_replay_episodes.json"
|
||
with open(replay_file, "w") as f:
|
||
json.dump(episodes, f)
|
||
|
||
log.info(f"[CL-RL] {len(episodes)} episodes reels sauvegardes -> {replay_file} "
|
||
f"(entrainement RL 100% reel -- pas de AhadQuantForexEnv)")
|
||
|
||
# Declencher fine_tune_real_only() via rl_train.py --replay (accept/
|
||
# reject integre, voir rl_train.py::fine_tune_real_only()). Pas de
|
||
# --steps : ce mode ne fait qu'un seul model.train() sur le buffer
|
||
# reel, donc largement plus rapide qu'un fine-tune simule.
|
||
import subprocess, sys
|
||
proc = subprocess.run(
|
||
[sys.executable, "rl_train.py",
|
||
"--finetune",
|
||
"--replay", replay_file,
|
||
"--real-min-trades", str(min_trades)],
|
||
capture_output=True, text=True, timeout=1800, # 30 min max -- pas de simulation, donc rapide
|
||
)
|
||
# Nettoyer le fichier temporaire dans tous les cas
|
||
if os.path.exists(replay_file):
|
||
os.remove(replay_file)
|
||
|
||
if proc.returncode != 0:
|
||
log.warning(f"[CL-RL] Echec fine-tune reel : {proc.stderr[-300:]}")
|
||
return False
|
||
|
||
accepted = False
|
||
for line in proc.stdout.splitlines():
|
||
if line.startswith("RL_FINETUNE_RESULT"):
|
||
accepted = "accepted=True" in line
|
||
break
|
||
|
||
if accepted:
|
||
log.info(f"[CL-RL] OK RL mis a jour -- {len(episodes)} vrais trades, aucune simulation")
|
||
try:
|
||
from rl_agent import reset_rl_agent
|
||
reset_rl_agent()
|
||
except ImportError:
|
||
pass
|
||
if notify_fn:
|
||
notify_fn(
|
||
f"OK *RL mis a jour (100% reel)*\n"
|
||
f"{len(episodes)} vrais trades integres au PPO -- aucune simulation"
|
||
)
|
||
return True
|
||
else:
|
||
log.info("[CL-RL] Fine-tune reel rejete -- politique conservee")
|
||
return False
|
||
|
||
except Exception as e:
|
||
log.error(f"[CL-RL] Erreur cycle RL reel : {e}")
|
||
return False
|
||
|
||
|
||
def _load_daily_local_state() -> dict:
|
||
if not os.path.exists(DAILY_LOCAL_TIMESTAMP_FILE):
|
||
return {"timestamp": 0.0, "datetime": "never"}
|
||
try:
|
||
with open(DAILY_LOCAL_TIMESTAMP_FILE) as f:
|
||
return json.load(f)
|
||
except Exception:
|
||
return {"timestamp": 0.0, "datetime": "never"}
|
||
|
||
|
||
def _save_daily_local_state():
|
||
with open(DAILY_LOCAL_TIMESTAMP_FILE, "w") as f:
|
||
json.dump({
|
||
"timestamp": time.time(),
|
||
"datetime": datetime.now(timezone.utc).isoformat(),
|
||
}, f, indent=2)
|
||
|
||
|
||
def should_run_daily_local_cycle() -> bool:
|
||
"""True si le cycle unifié quotidien et local (ML + RL) est dû."""
|
||
if not getattr(config, "DAILY_LOCAL_RETRAIN_ENABLED", True):
|
||
return False
|
||
interval = getattr(config, "DAILY_LOCAL_RETRAIN_INTERVAL_HOURS", 24) * 3600
|
||
elapsed = time.time() - _load_daily_local_state().get("timestamp", 0.0)
|
||
return elapsed >= interval
|
||
|
||
|
||
def run_daily_unified_retrain(notify_fn=None) -> dict:
|
||
"""
|
||
Cycle d'entraînement UNIFIÉ, quotidien et 100% LOCAL pour ML et RL.
|
||
[FIX] Protégé par _retrain_lock — ne peut pas tourner en même temps
|
||
qu'un full-retrain ou qu'un autre cycle local.
|
||
"""
|
||
if not _retrain_lock.acquire(blocking=False):
|
||
log.warning("[DAILY-LOCAL] Un autre cycle d'entraînement est déjà en cours — skip")
|
||
return {"skipped": True, "reason": "retrain_lock"}
|
||
try:
|
||
return _run_daily_unified_retrain_inner(notify_fn)
|
||
finally:
|
||
_retrain_lock.release()
|
||
|
||
|
||
def _run_daily_unified_retrain_inner(notify_fn=None) -> dict:
|
||
"""
|
||
Cycle d'entraînement UNIFIÉ, quotidien et 100% LOCAL pour ML et RL —
|
||
traités comme un seul et même modèle qui se met à jour ensemble.
|
||
|
||
Aucune des deux étapes ne télécharge quoi que ce soit sur le réseau :
|
||
1. ML — warm-start LightGBM EMBARQUÉ DANS L'ENSEMBLE
|
||
(online_learner.py::daily_update, sur les trades réels du
|
||
buffer local des dernières 24h). N'écrase model_ensemble.pkl
|
||
que si l'accuracy de l'ENSEMBLE COMPLET s'améliore (sinon
|
||
l'ancien est conservé) — point #3 du diagnostic.
|
||
2. RL — [21/06/2026] désormais 100% basé sur les vrais trades du
|
||
bot, ZÉRO simulation (voir run_rl_retrain_with_replay et
|
||
rl_train.py::fine_tune_real_only). N'écrase rl_agent.zip que
|
||
si la nouvelle politique obtient un reward moyen au moins
|
||
équivalent à l'ancienne (yardstick simulé, comparaison
|
||
uniquement). Si moins de RL_REPLAY_MIN_TRADES trades réels
|
||
sont disponibles, cette étape est SKIP — pas de repli sur la
|
||
simulation.
|
||
|
||
Si l'une des deux étapes a effectivement remplacé un modèle, le bundle
|
||
portable ahad_quant_unified.zip est régénéré (best-effort) pour rester
|
||
cohérent avec les fichiers vivants.
|
||
|
||
Retourne {"ml_updated": bool, "rl_updated": bool, "ran": bool}.
|
||
"""
|
||
result = {"ml_updated": False, "rl_updated": False, "ran": False}
|
||
|
||
if not getattr(config, "DAILY_LOCAL_RETRAIN_ENABLED", True):
|
||
log.debug("[DAILY-LOCAL] Désactivé (DAILY_LOCAL_RETRAIN_ENABLED=false)")
|
||
return result
|
||
|
||
result["ran"] = True
|
||
log.info("[DAILY-LOCAL] === Démarrage du cycle quotidien unifié ML+RL (local) ===")
|
||
|
||
# ── 1. ML — warm-start local de l'ensemble (accept/reject intégré) ─────
|
||
try:
|
||
ml_updated = run_daily_lightgbm_warmstart(notify_fn=notify_fn)
|
||
result["ml_updated"] = bool(ml_updated)
|
||
except Exception as e:
|
||
log.error(f"[DAILY-LOCAL] Erreur étape ML : {e}")
|
||
|
||
# ── 2. RL — fine-tune local + replay réel (accept/reject intégré) ──────
|
||
if getattr(config, "RL_AUTO_RETRAIN_ENABLED", True):
|
||
try:
|
||
rl_updated = run_rl_retrain_with_replay(notify_fn=notify_fn)
|
||
result["rl_updated"] = bool(rl_updated)
|
||
except Exception as e:
|
||
log.error(f"[DAILY-LOCAL] Erreur étape RL : {e}")
|
||
else:
|
||
log.debug("[DAILY-LOCAL] RL désactivé (RL_AUTO_RETRAIN_ENABLED=false)")
|
||
|
||
# ── 3. Bilan ─────────────────────────────────────────────────────────
|
||
_save_daily_local_state()
|
||
|
||
if result["ml_updated"] or result["rl_updated"]:
|
||
parts = []
|
||
if result["ml_updated"]: parts.append("ML mis à jour")
|
||
if result["rl_updated"]: parts.append("RL mis à jour")
|
||
msg = f"🔄 *Cycle quotidien local terminé* — {', '.join(parts)}"
|
||
log.info(msg.replace("*", ""))
|
||
if notify_fn:
|
||
notify_fn(msg)
|
||
else:
|
||
log.info("[DAILY-LOCAL] === Cycle terminé — aucun modèle n'a été amélioré, anciens modèles conservés ===")
|
||
|
||
return result
|
||
"""
|
||
Vérification horaire du monitor de performance.
|
||
Retourne le statut (OK/WARNING/DANGER/EMERGENCY).
|
||
"""
|
||
if not getattr(config, "CONTINUOUS_LEARNING_ENABLED", False):
|
||
return "OK"
|
||
if not getattr(config, "MONITOR_ENABLED", False):
|
||
return "OK"
|
||
try:
|
||
from experience_buffer import get_experience_buffer
|
||
from performance_monitor import PerformanceMonitor
|
||
|
||
buf = get_experience_buffer(auto_load=False)
|
||
monitor = PerformanceMonitor(buf)
|
||
return monitor.check()
|
||
except Exception as e:
|
||
log.debug(f"[CL-MONITOR] Erreur check horaire : {e}")
|
||
return "OK"
|