#!/usr/bin/env python3 """ 🔧 SPY PARAM PATCHER — Application automatique des recommandations spy_exit_analyzer Lit spy_exit_analysis.json et propose des ajustements de paramètres dans market_spy.py basés sur les statistiques contrefactuelles mesurées sur les trades réels. Chaque patch est: - Justifié par des données chiffrées - Présenté avec avant/après - Appliqué seulement après confirmation interactive - Appliqué au testnet d'abord, prod séparément Usage: python3 spy_param_patcher.py # Mode interactif python3 spy_param_patcher.py --dry-run # Affiche les patches sans appliquer python3 spy_param_patcher.py --auto # Applique toutes les suggestions HIGH priority sans confirmation """ import argparse import json import re import shutil from datetime import datetime, timezone from pathlib import Path # ─── CONFIG ──────────────────────────────────────────────────────────────────── ANALYSIS_FILE = Path("/home/ubuntu/crypto_trading_bot/spy_exit_analysis.json") TESTNET_SCRIPT = Path("/home/ubuntu/crypto_trading_bot/market_spy.py") PROD_SCRIPT = Path("/home/ubuntu/crypto_trading_prod/market_spy.py") PATCH_LOG = Path("/home/ubuntu/crypto_trading_bot/spy_patch_history.json") # ─── DÉFINITIONS DES PATCHES ─────────────────────────────────────────────────── # Chaque patch = règle qui le déclenche + conditions + modification à effectuer PATCH_CATALOG = [ { "id": "EARLY_SL_DELAY", "priority": "HIGH", "trigger_rule": "EARLY_SL", "condition": lambda stats: stats.get("wicks", 0) / max(stats.get("candidates", 1), 1) > 0.60, "description": "EARLY_SL: {wick_rate:.0%} des cas sont des creux temporaires → délai trop court", "param": "EARLY_SL_DELAY", "pattern": r"(EARLY_SL_DELAY\s*=\s*)(\d+)", "compute": lambda current, stats: min(current + 30, 150), # +30s, max 2.5min "unit": "secondes", "rationale": "Laisser 30s de plus avant de déclencher → réduit les faux positifs", }, { "id": "EARLY_SL_MIN_PNL", "priority": "HIGH", "trigger_rule": "EARLY_SL", "condition": lambda stats: stats.get("avg_gain_missed", 0) > 2.0 and stats.get("wicks", 0) / max(stats.get("candidates", 1), 1) > 0.70, "description": "EARLY_SL: gain moy. manqué {avg_gain:.2f}% → exiger un min_pnl plus bas pour tirer le frein", "param": "EARLY_SL_MIN_PNL", "pattern": r"(EARLY_SL_MIN_PNL\s*=\s*)([\d.]+)", "compute": lambda current, stats: round(max(current - 0.05, 0.10), 2), # -0.05%, floor 0.10% "unit": "%", "rationale": "Si le prix n'a même pas atteint 0.15% → signal moins douteux que si 0.20% jamais touché", }, { "id": "MOMENTUM_EXIT_CANDLES", "priority": "HIGH", "trigger_rule": "MOMENTUM_EXIT", "condition": lambda stats: stats.get("wicks", 0) / max(stats.get("candidates", 1), 1) > 0.70, "description": "MOMENTUM_EXIT: {wick_rate:.0%} des sorties étaient des creux → 3 baisses trop sensible", "param": "MOMENTUM_EXIT_CANDLES", "pattern": r"(MOMENTUM_EXIT_CANDLES\s*=\s*)(\d+)", "compute": lambda current, stats: min(current + 1, 5), # +1 candle, max 5 "unit": "baisses consécutives", "rationale": "Passer de 3 à 4 baisses consécutives réduit les faux retournements", }, { "id": "STAGNATION_EXIT_MINUTES", "priority": "MED", "trigger_rule": "STAGNATION", "condition": lambda stats: stats.get("avg_gain_missed", 0) > 1.0 and stats.get("wicks", 0) / max(stats.get("candidates", 1), 1) > 0.50, "description": "STAGNATION: {wick_rate:.0%} des sorties avaient un rebond après → délai trop court", "param": "STAGNATION_EXIT_MINUTES", "pattern": r"(STAGNATION_EXIT_MINUTES\s*=\s*)(\d+)", "compute": lambda current, stats: min(current + 2, 20), # +2min, max 20min "unit": "minutes", "rationale": "Attendre 2min de plus avant sortie stagnation — range consolidation classique", }, { "id": "TRAILING_STOP_WIDE", "priority": "MED", "trigger_rule": "TRAILING", "condition": lambda stats: stats.get("big_gap_rate", 0) > 0.30, "description": "TRAILING: {big_gap_rate:.0%} des trades ont laissé >2% sur la table", "param": "TRAILING_STOP_WIDE", "pattern": r"(TRAILING_STOP_WIDE\s*=\s*)([\d.]+)", "compute": lambda current, stats: round(min(current + 0.3, 2.5), 1), # +0.3%, max 2.5% "unit": "%", "rationale": "Élargir trailing moyen (2-5% range) de 1.5→1.8% → laisser courir les bonnes tendances", }, ] # ─── LECTURE ANALYSE ─────────────────────────────────────────────────────────── def load_analysis() -> dict: if not ANALYSIS_FILE.exists(): print(f"❌ Fichier d'analyse introuvable: {ANALYSIS_FILE}") print(" Lancez d'abord: python3 spy_exit_analyzer.py") return {} with open(ANALYSIS_FILE, encoding="utf-8") as f: return json.load(f) def compute_trailing_big_gap_rate() -> float: """Calcule le taux de trades TRAILING qui ont laissé >2% sur la table.""" try: hist = Path("/home/ubuntu/crypto_trading_prod/data/espion_history.json") with open(hist, encoding="utf-8") as f: trades = json.load(f) trail = [t for t in trades if "TRAILING" in t.get("exit_reason", "")] if not trail: return 0.0 big_gap = [t for t in trail if t.get("max_pnl", 0) - t.get("pnl_pct", 0) > 2.0] return len(big_gap) / len(trail) except Exception: return 0.0 # ─── LECTURE VALEUR ACTUELLE ─────────────────────────────────────────────────── def read_current_value(script_path: Path, pattern: str) -> tuple[float | None, str]: """Lit la valeur actuelle d'un paramètre dans le script.""" content = script_path.read_text(encoding="utf-8") m = re.search(pattern, content) if not m: return None, content try: val = float(m.group(2)) except ValueError: val = int(m.group(2)) return val, content # ─── APPLICATION DU PATCH ────────────────────────────────────────────────────── def apply_patch(script_path: Path, patch: dict, new_value: float | int, dry_run: bool) -> bool: """Applique un patch à un script. Retourne True si succès.""" current, content = read_current_value(script_path, patch["pattern"]) if current is None: print(f" ⚠️ Paramètre {patch['param']} non trouvé dans {script_path.name}") return False # Formater la valeur if isinstance(new_value, float) and new_value != int(new_value): new_str = str(new_value) else: new_str = str(int(new_value)) # Construire le remplacement def replacer(m): return m.group(1) + new_str new_content = re.sub(patch["pattern"], replacer, content, count=1) if new_content == content: print(f" ⚠️ Aucun changement effectué (pattern non trouvé)") return False if dry_run: print(f" [DRY-RUN] {current} → {new_value}") return True # Backup ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") backup = script_path.with_suffix(f".py.bak_{ts}") shutil.copy2(script_path, backup) # Écriture script_path.write_text(new_content, encoding="utf-8") print(f" ✅ Patch appliqué: {current} → {new_value} ({script_path.name})") print(f" Backup: {backup.name}") return True # ─── POINT D'ENTRÉE ──────────────────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser(description="Patch automatique des paramètres spy basé sur l'analyse contrefactuelle") parser.add_argument("--dry-run", action="store_true", help="Affiche les patches sans les appliquer") parser.add_argument("--auto", action="store_true", help="Applique les patches HIGH priority sans confirmation") args = parser.parse_args() print("\n" + "═" * 60) print(" 🔧 SPY PARAM PATCHER") print(" Basé sur: spy_exit_analysis.json") print("═" * 60) analysis = load_analysis() if not analysis: return ts = analysis.get("generated_at", "?") n_trades = analysis.get("total_trades", 0) n_cands = analysis.get("candidates_analyzed", 0) avg_miss = analysis.get("avg_gain_missed", 0) print(f"\n Analyse du: {ts[:19].replace('T', ' ')} UTC") print(f" {n_trades} trades | {n_cands} candidats analysés | gain moy. manqué: {avg_miss:+.2f}%\n") by_rule = analysis.get("by_rule_summary", {}) # Enrichir TRAILING avec big_gap_rate trail_big_gap = compute_trailing_big_gap_rate() patches_to_apply = [] print("📋 PATCHES DISPONIBLES:") print("─" * 60) for patch in PATCH_CATALOG: rule = patch["trigger_rule"] # Stats pour ce patch if rule == "TRAILING": stats = {"big_gap_rate": trail_big_gap} else: stats = by_rule.get(rule, {}) if stats: stats["wick_rate"] = stats.get("wicks", 0) / max(stats.get("candidates", 1), 1) if not stats: continue # Vérifier la condition if not patch["condition"](stats): continue # Lire valeur actuelle (testnet comme référence) current, _ = read_current_value(TESTNET_SCRIPT, patch["pattern"]) if current is None: continue # Calculer nouvelle valeur new_value = patch["compute"](current, stats) if new_value == current: continue # Formater la description wick_rate = stats.get("wick_rate", stats.get("big_gap_rate", 0)) avg_gain = stats.get("avg_gain_missed", 0) desc = patch["description"].format( wick_rate=wick_rate, avg_gain=avg_gain, big_gap_rate=stats.get("big_gap_rate", 0) ) prio_icon = "🔴" if patch["priority"] == "HIGH" else "🟡" print(f"\n [{prio_icon} {patch['priority']}] {patch['param']}") print(f" Motif : {desc}") print(f" Avant : {current} {patch['unit']}") print(f" Après : {new_value} {patch['unit']}") print(f" Pourquoi: {patch['rationale']}") patches_to_apply.append({ "patch": patch, "current": current, "new_value": new_value, "stats": stats, }) if not patches_to_apply: print(" ✅ Aucun patch nécessaire — paramètres déjà optimaux pour les données actuelles.") return print(f"\n{'─' * 60}") print(f" {len(patches_to_apply)} patch(s) identifié(s)") if args.dry_run: print("\n [MODE DRY-RUN — aucune modification]\n") return # ── Confirmation et application ── patch_log = [] for entry in patches_to_apply: patch = entry["patch"] current = entry["current"] new_value = entry["new_value"] prio = patch["priority"] print(f"\n─── {patch['param']}: {current} → {new_value} ───") if args.auto and prio == "HIGH": do_apply = True print(" [--auto] Application automatique (HIGH priority)") else: resp = input(f" Appliquer ce patch ? [o/N] ").strip().lower() do_apply = resp in ("o", "oui", "y", "yes") if not do_apply: print(" ⏭️ Skipped") continue # Testnet print(f"\n → TESTNET ({TESTNET_SCRIPT.name}):") ok_test = apply_patch(TESTNET_SCRIPT, patch, new_value, dry_run=False) if not ok_test: continue # Prod — confirmation séparée resp_prod = input(f" → Appliquer aussi en PRODUCTION ? [o/N] ").strip().lower() if resp_prod in ("o", "oui", "y", "yes"): print(f" → PROD ({PROD_SCRIPT.name}):") apply_patch(PROD_SCRIPT, patch, new_value, dry_run=False) else: print(" ⏭️ Prod skipped") # Journal patch_log.append({ "timestamp": datetime.now(timezone.utc).isoformat(), "patch_id": patch["id"], "param": patch["param"], "old_value": current, "new_value": new_value, "trigger_rule": patch["trigger_rule"], "analysis_date": analysis.get("generated_at"), }) # ── Sauvegarder le journal ── if patch_log: history = [] if PATCH_LOG.exists(): with open(PATCH_LOG, encoding="utf-8") as f: history = json.load(f) history.extend(patch_log) with open(PATCH_LOG, "w", encoding="utf-8") as f: json.dump(history, f, indent=2, ensure_ascii=False) print(f"\n✅ {len(patch_log)} patch(s) appliqué(s) — journal: {PATCH_LOG.name}") print(" ⚠️ Redémarrez le bot pour prendre en compte les changements.") else: print("\n Aucun patch appliqué.") if __name__ == "__main__": main()