ahad-quant/mt5_bridge.py

"""
AHAD QUANT — MT5 Bridge (CSV File Communication)
================================================
Assure la communication temps réel entre AHAD QUANT (Python) et un EA MetaTrader 5
via des fichiers CSV partagés dans le dossier MQL5/Files/.

Flux :
  1. AHAD QUANT écrit un signal dans signals.csv
  2. L'EA MT5 lit signals.csv via OnTimer(), exécute l'ordre
  3. L'EA écrit le résultat dans reports.csv
  4. MT5Bridge détecte le changement (watchdog <50ms) et appelle le callback
  5. L'EA écrit status.csv toutes les 30s (balance, equity, positions)

Usage :
    from mt5_bridge import MT5Bridge

    bridge = MT5Bridge(files_path="C:/Users/NOM/.../MQL5/Files")

    bridge.on_report_received(lambda r: print("Rapport reçu:", r))
    bridge.on_status_updated(lambda s: print("Statut compte:", s))

    bridge.start()

    bridge.write_signal("EURUSD", "BUY", lot=0.1, sl_pips=30, tp_pips=60, confidence=0.82)
    bridge.write_signal("GBPUSD", "SELL", lot=0.05, sl_pips=25, tp_pips=50, confidence=0.75)

    bridge.stop()

Dépendances :
    pip install watchdog filelock
"""

import csv
import logging
import os
import threading
import time
import uuid
from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Callable, List, Optional

try:
    from watchdog.events import FileSystemEventHandler, FileModifiedEvent
    from watchdog.observers import Observer
    HAS_WATCHDOG = True
except ImportError:
    HAS_WATCHDOG = False
    print("[MT5Bridge] AVERTISSEMENT : watchdog non installé. Utilisation du polling 200ms.")
    print("[MT5Bridge] Pour installer : pip install watchdog")

try:
    from filelock import FileLock, Timeout as FileLockTimeout
    HAS_FILELOCK = True
except ImportError:
    HAS_FILELOCK = False
    print("[MT5Bridge] AVERTISSEMENT : filelock non installé.")
    print("[MT5Bridge] Pour installer : pip install filelock")

logger = logging.getLogger(__name__)


# ─── Structures de données ────────────────────────────────────────────────────

@dataclass
class Signal:
    """Un signal de trading à envoyer à l'EA MT5."""
    pair: str               # Ex: "EURUSD"
    action: str             # "BUY" ou "SELL"
    lot_size: float         # Volume en lots (ex: 0.1)
    sl_pips: int            # Stop loss en pips (ex: 30)
    tp_pips: int            # Take profit en pips (ex: 60)
    confidence: float       # Score de confiance du modèle (0.0 à 1.0)
    signal_id: str = field(default_factory=lambda: f"SIG_{uuid.uuid4().hex[:8].upper()}")
    timestamp: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())


@dataclass
class TradeReport:
    """Un rapport de trade renvoyé par l'EA MT5."""
    signal_id: str
    ticket: int             # Numéro de ticket MT5
    status: str             # "OPEN", "CLOSED", "ERROR", "REJECTED"
    open_price: float
    sl: float
    tp: float
    open_time: str
    close_price: Optional[float] = None
    close_time: Optional[str] = None
    profit: Optional[float] = None
    comment: Optional[str] = None


@dataclass
class AccountStatus:
    """État du compte MT5 mis à jour toutes les 30s par l'EA."""
    timestamp: str
    balance: float
    equity: float
    margin: float
    free_margin: float
    open_positions: int
    daily_pnl: float


# ─── Watcher de fichier (watchdog ou polling) ─────────────────────────────────

class _CSVChangeHandler(FileSystemEventHandler):
    """Déclenche un callback dès qu'un fichier CSV est modifié."""

    def __init__(self, filepath: str, callback: Callable):
        super().__init__()
        self._filepath = os.path.abspath(filepath)
        self._callback = callback

    def on_modified(self, event):
        if not event.is_directory and os.path.abspath(event.src_path) == self._filepath:
            self._callback()


class _PollingWatcher:
    """Fallback si watchdog n'est pas installé — poll toutes les 200ms."""

    def __init__(self, filepath: str, callback: Callable, interval: float = 0.2):
        self._filepath = filepath
        self._callback = callback
        self._interval = interval
        self._last_mtime: float = 0.0
        self._running = False
        self._thread: Optional[threading.Thread] = None

    def start(self):
        self._running = True
        self._thread = threading.Thread(target=self._loop, daemon=True)
        self._thread.start()

    def stop(self):
        self._running = False

    def _loop(self):
        while self._running:
            try:
                mtime = os.path.getmtime(self._filepath) if os.path.exists(self._filepath) else 0
                if mtime > self._last_mtime:
                    self._last_mtime = mtime
                    if mtime > 0:
                        self._callback()
            except OSError:
                pass
            time.sleep(self._interval)


# ─── Classe principale MT5Bridge ──────────────────────────────────────────────

class MT5Bridge:
    """
    Pont de communication temps réel entre AHAD QUANT (Python) et un EA MetaTrader 5.

    Paramètres
    ----------
    files_path : str
        Chemin absolu vers le dossier MQL5/Files/ de MetaTrader 5.
        Ex Windows : "C:/Users/NOM/AppData/Roaming/MetaQuotes/Terminal/XXXXX/MQL5/Files"

    signal_file : str
        Nom du fichier de signaux (défaut : "signals.csv")

    report_file : str
        Nom du fichier de rapports (défaut : "reports.csv")

    status_file : str
        Nom du fichier de statut (défaut : "status.csv")

    signal_timeout : int
        Délai en secondes après lequel un signal non exécuté est considéré périmé (défaut : 30)
    """

    # En-têtes des fichiers CSV
    _SIGNAL_HEADERS  = ["timestamp", "pair", "action", "lot_size", "sl_pips",
                        "tp_pips", "signal_id", "confidence"]
    _REPORT_HEADERS  = ["signal_id", "ticket", "status", "open_price", "sl", "tp",
                        "open_time", "close_price", "close_time", "profit", "comment"]
    _STATUS_HEADERS  = ["timestamp", "balance", "equity", "margin",
                        "free_margin", "open_positions", "daily_pnl"]

    def __init__(
        self,
        files_path: str,
        signal_file: str = "signals.csv",
        report_file: str = "reports.csv",
        status_file: str = "status.csv",
        signal_timeout: int = 30,
        max_signal_rows: int = 500,
        signal_rotate_check_every: int = 25,
    ):
        self._dir = Path(files_path)
        self._signal_path = self._dir / signal_file
        self._report_path = self._dir / report_file
        self._status_path = self._dir / status_file
        self._signal_timeout = signal_timeout
        self._max_signal_rows = max_signal_rows
        self._signal_rotate_check_every = signal_rotate_check_every
        self._signal_write_count = 0

        # Callbacks enregistrés par l'utilisateur
        self._report_callbacks:  List[Callable[[TradeReport], None]] = []
        self._status_callbacks:  List[Callable[[AccountStatus], None]] = []

        # IDs de rapports déjà traités (évite les doublons)
        self._seen_report_ids: set = set()

        # Watchers
        self._report_observer = None
        self._status_observer = None
        self._running = False
        self._lock = threading.Lock()

    # ─── API publique ──────────────────────────────────────────────────────────

    def on_report_received(self, callback: Callable[[TradeReport], None]) -> None:
        """Enregistre un callback appelé à chaque nouveau rapport de trade."""
        self._report_callbacks.append(callback)

    def on_status_updated(self, callback: Callable[[AccountStatus], None]) -> None:
        """Enregistre un callback appelé à chaque mise à jour du statut compte."""
        self._status_callbacks.append(callback)

    def start(self) -> None:
        """
        Démarre le bridge : initialise les fichiers CSV et lance les watchers.
        À appeler une fois au démarrage du bot.
        """
        self._dir.mkdir(parents=True, exist_ok=True)
        self._init_signal_file()
        self._init_report_file()

        self._running = True
        self._start_watcher(self._report_path, self._on_report_file_changed, "report")
        self._start_watcher(self._status_path, self._on_status_file_changed, "status")

        logger.info(f"[MT5Bridge] Démarré — dossier : {self._dir}")
        logger.info(f"[MT5Bridge] signals  → {self._signal_path}")
        logger.info(f"[MT5Bridge] reports  ← {self._report_path}")
        logger.info(f"[MT5Bridge] status   ← {self._status_path}")

    def stop(self) -> None:
        """Arrête proprement les watchers."""
        self._running = False
        if self._report_observer:
            try:
                self._report_observer.stop()
                self._report_observer.join(timeout=2)
            except Exception:
                pass
        if self._status_observer:
            try:
                self._status_observer.stop()
                self._status_observer.join(timeout=2)
            except Exception:
                pass
        logger.info("[MT5Bridge] Arrêté.")

    def write_signal(
        self,
        pair: str,
        action: str,
        lot: float,
        sl_pips: int,
        tp_pips: int,
        confidence: float,
    ) -> Signal:
        """
        Écrit un signal de trading dans signals.csv pour l'EA MT5.

        Paramètres
        ----------
        pair       : Paire Forex — ex: "EURUSD"
        action     : "BUY" ou "SELL"
        lot        : Volume en lots — ex: 0.1
        sl_pips    : Stop loss en pips — ex: 30
        tp_pips    : Take profit en pips — ex: 60
        confidence : Score modèle entre 0 et 1 — ex: 0.82

        Retourne
        --------
        Signal : l'objet signal créé, avec son signal_id unique
        """
        action = action.upper()
        if action not in ("BUY", "SELL"):
            raise ValueError(f"action doit être 'BUY' ou 'SELL', reçu : '{action}'")
        if not (0.0 < lot <= 100.0):
            raise ValueError(f"lot invalide : {lot}")
        if sl_pips <= 0 or tp_pips <= 0:
            raise ValueError(f"sl_pips et tp_pips doivent être > 0")

        # Bug #25 fix : ne pas uppercaser la paire entière.
        # pair.upper() convertit "EURCHF.m" → "EURCHF.M" (M majuscule) alors que
        # le broker utilise le suffixe minuscule ".m" → SYMBOL_NOT_FOUND côté EA.
        # La paire de base (ex: "EURCHF") est déjà en majuscules côté appelant.
        sig = Signal(
            pair=pair,
            action=action,
            lot_size=lot,
            sl_pips=sl_pips,
            tp_pips=tp_pips,
            confidence=round(confidence, 4),
        )

        self._append_to_csv(self._signal_path, self._SIGNAL_HEADERS, asdict(sig))
        logger.info(f"[MT5Bridge] Signal écrit : {sig.signal_id} | {sig.pair} {sig.action} "
                    f"lot={sig.lot_size} SL={sig.sl_pips}p TP={sig.tp_pips}p conf={sig.confidence}")
        self._signal_write_count += 1
        if self._signal_write_count % self._signal_rotate_check_every == 0:
            self._rotate_signal_file_if_needed()
        return sig

    def read_reports(self) -> List[TradeReport]:
        """
        Lit tous les rapports présents dans reports.csv.
        Retourne uniquement les rapports nouveaux (non encore traités).
        """
        return self._read_new_reports()

    def read_status(self) -> Optional[AccountStatus]:
        """
        Lit le dernier statut du compte depuis status.csv.
        Retourne None si le fichier n'existe pas encore.
        """
        return self._read_latest_status()

    def get_open_signals(self) -> List[dict]:
        """
        Retourne les signaux envoyés à l'EA qui n'ont pas encore reçu de rapport.
        Filtre les signaux périmés (plus vieux que signal_timeout secondes).
        """
        if not self._signal_path.exists():
            return []

        now = datetime.now(timezone.utc).timestamp()
        open_sigs = []

        with self._safe_read(self._signal_path) as rows:
            for row in rows:
                try:
                    ts = datetime.fromisoformat(row["timestamp"]).timestamp()
                    if now - ts <= self._signal_timeout:
                        open_sigs.append(row)
                except (KeyError, ValueError):
                    continue

        return open_sigs

    # ─── Initialisation des fichiers ──────────────────────────────────────────

    def _init_signal_file(self) -> None:
        """Crée signals.csv avec les en-têtes s'il n'existe pas."""
        if not self._signal_path.exists():
            self._write_headers(self._signal_path, self._SIGNAL_HEADERS)
            logger.info(f"[MT5Bridge] signals.csv créé : {self._signal_path}")

    def _init_report_file(self) -> None:
        """Crée reports.csv avec les en-têtes s'il n'existe pas."""
        if not self._report_path.exists():
            self._write_headers(self._report_path, self._REPORT_HEADERS)
            logger.info(f"[MT5Bridge] reports.csv créé : {self._report_path}")

    def _rotate_signal_file_if_needed(self) -> None:
        """
        Tronque signals.csv aux _max_signal_rows lignes les plus récentes.
        Les signaux non encore acquittés (absents de reports.csv) sont TOUJOURS
        conservés, même s'ils sont hors de la fenêtre des dernières lignes.
        Protégé par FileLock pour cohérence avec le reste du bridge.
        """
        if not self._signal_path.exists():
            return
        try:
            lock_path = str(self._signal_path) + ".lock"
            ctx = FileLock(lock_path, timeout=3) if HAS_FILELOCK else None

            def _do_rotate():
                with open(self._signal_path, encoding="utf-8-sig", newline="") as fh:
                    reader = csv.DictReader(fh)
                    rows = list(reader)
                    headers = reader.fieldnames or self._SIGNAL_HEADERS

                if len(rows) <= self._max_signal_rows:
                    return  # Pas assez de lignes — rien à faire

                # Récupérer les signal_ids déjà ACK dans reports.csv
                ack_ids: set = set()
                if self._report_path.exists():
                    with open(self._report_path, encoding="utf-8-sig", newline="") as rh:
                        for r in csv.DictReader(rh):
                            sid = r.get("signal_id")
                            if sid:
                                ack_ids.add(sid)

                # Garder les N dernières lignes + toutes les lignes non ACK hors fenêtre
                tail = rows[-self._max_signal_rows:]
                tail_ids = {r.get("signal_id") for r in tail}
                pending_outside = [r for r in rows[:-self._max_signal_rows]
                                   if r.get("signal_id") not in ack_ids
                                   and r.get("signal_id") not in tail_ids]
                kept = pending_outside + tail

                # Réécriture atomique via fichier temporaire
                tmp = str(self._signal_path) + ".tmp"
                with open(tmp, "w", encoding="utf-8", newline="") as fh:
                    writer = csv.DictWriter(fh, fieldnames=headers)
                    writer.writeheader()
                    writer.writerows(kept)
                os.replace(tmp, self._signal_path)
                logger.info(f"[MT5Bridge] signals.csv rotation : {len(rows)} → {len(kept)} lignes"
                            f" ({len(pending_outside)} signaux pending conservés hors fenêtre)")

            if ctx:
                with ctx:
                    _do_rotate()
            else:
                _do_rotate()

        except Exception as exc:
            logger.warning(f"[MT5Bridge] Rotation signals.csv échouée : {exc}")

    # ─── Watchers ─────────────────────────────────────────────────────────────

    def _start_watcher(self, filepath: Path, callback: Callable, name: str) -> None:
        """Lance un watcher watchdog ou polling selon ce qui est disponible."""
        if HAS_WATCHDOG:
            handler = _CSVChangeHandler(str(filepath), callback)
            observer = Observer()
            observer.schedule(handler, path=str(filepath.parent), recursive=False)
            observer.start()
            if name == "report":
                self._report_observer = observer
            else:
                self._status_observer = observer
            logger.debug(f"[MT5Bridge] Watcher watchdog démarré pour {name}")
        else:
            watcher = _PollingWatcher(str(filepath), callback)
            watcher.start()
            if name == "report":
                self._report_observer = watcher
            else:
                self._status_observer = watcher
            logger.debug(f"[MT5Bridge] Watcher polling démarré pour {name}")

    # ─── Callbacks de détection de changement ─────────────────────────────────

    def _on_report_file_changed(self) -> None:
        """Appelé dès que reports.csv est modifié."""
        time.sleep(0.02)  # Laisser l'EA finir d'écrire
        new_reports = self._read_new_reports()
        for report in new_reports:
            logger.info(f"[MT5Bridge] Rapport reçu : {report.signal_id} | "
                        f"ticket={report.ticket} statut={report.status} "
                        f"profit={report.profit}")
            for cb in self._report_callbacks:
                try:
                    cb(report)
                except Exception as e:
                    logger.error(f"[MT5Bridge] Erreur callback rapport : {e}")

    def _on_status_file_changed(self) -> None:
        """Appelé dès que status.csv est modifié."""
        time.sleep(0.02)
        status = self._read_latest_status()
        if status:
            logger.debug(f"[MT5Bridge] Statut compte : balance={status.balance} "
                         f"equity={status.equity} positions={status.open_positions}")
            for cb in self._status_callbacks:
                try:
                    cb(status)
                except Exception as e:
                    logger.error(f"[MT5Bridge] Erreur callback statut : {e}")

    # ─── Lecture des fichiers ─────────────────────────────────────────────────

    def _read_new_reports(self) -> List[TradeReport]:
        """Lit reports.csv et retourne uniquement les lignes non encore traitées."""
        if not self._report_path.exists():
            return []

        new_reports = []
        with self._safe_read(self._report_path) as rows:
            for row in rows:
                try:
                    # Clé unique : signal_id + statut (un OPEN et un CLOSED pour le même signal)
                    key = f"{row['signal_id']}_{row['status']}"
                    if key in self._seen_report_ids:
                        continue
                    self._seen_report_ids.add(key)

                    report = TradeReport(
                        signal_id   = row["signal_id"],
                        ticket      = int(row["ticket"]) if row.get("ticket") else 0,
                        status      = row["status"],
                        open_price  = float(row["open_price"]) if row.get("open_price") else 0.0,
                        sl          = float(row["sl"]) if row.get("sl") else 0.0,
                        tp          = float(row["tp"]) if row.get("tp") else 0.0,
                        open_time   = row.get("open_time", ""),
                        close_price = float(row["close_price"]) if row.get("close_price") else None,
                        close_time  = row.get("close_time") or None,
                        profit      = float(row["profit"]) if row.get("profit") else None,
                        comment     = row.get("comment") or None,
                    )
                    new_reports.append(report)
                except (KeyError, ValueError, TypeError) as e:
                    # Marquer la ligne comme vue pour éviter le spam toutes les 200ms
                    bad_key = f"_BAD_{hash(str(sorted(row.items())))}"
                    if bad_key not in self._seen_report_ids:
                        self._seen_report_ids.add(bad_key)
                        logger.warning(f"[MT5Bridge] Ligne rapport ignorée (format invalide) : {e}")

        return new_reports

    def _read_latest_status(self) -> Optional[AccountStatus]:
        """Lit la dernière ligne de status.csv (la plus récente)."""
        if not self._status_path.exists():
            return None

        with self._safe_read(self._status_path) as rows:
            valid_rows = []
            for row in rows:
                try:
                    valid_rows.append(AccountStatus(
                        timestamp      = row["timestamp"],
                        balance        = float(row["balance"]),
                        equity         = float(row["equity"]),
                        margin         = float(row["margin"]),
                        free_margin    = float(row["free_margin"]),
                        open_positions = int(row["open_positions"]),
                        daily_pnl      = float(row["daily_pnl"]),
                    ))
                except (KeyError, ValueError, TypeError):
                    continue

        return valid_rows[-1] if valid_rows else None

    # ─── Utilitaires CSV ──────────────────────────────────────────────────────

    def _write_headers(self, filepath: Path, headers: List[str]) -> None:
        """Crée un fichier CSV avec uniquement la ligne d'en-tête."""
        lock_path = str(filepath) + ".lock"
        if HAS_FILELOCK:
            with FileLock(lock_path, timeout=5):
                self._open_with_retry(filepath, "w", headers, row=None)
        else:
            self._open_with_retry(filepath, "w", headers, row=None)

    def _append_to_csv(self, filepath: Path, headers: List[str], row: dict) -> None:
        """
        Ajoute une ligne à un fichier CSV existant (thread-safe).

        Bug #24 fix : l'EA MT5 peut maintenir un lock Windows exclusif sur
        signals.csv pendant sa lecture (timer 200ms). Le open() Python reçoit
        alors PermissionError — non géré par FileLock (qui gère uniquement les
        conflits Python↔Python via un .lock séparé).
        Solution : retry avec backoff exponentiel (max 5 tentatives, 50ms base).
        """
        lock_path = str(filepath) + ".lock"
        if HAS_FILELOCK:
            try:
                with FileLock(lock_path, timeout=5):
                    self._open_with_retry(filepath, "a", headers, row)
            except FileLockTimeout:
                logger.error(f"[MT5Bridge] Impossible d'obtenir le verrou pour {filepath.name}")
        else:
            with self._lock:
                self._open_with_retry(filepath, "a", headers, row)

    def _open_with_retry(
        self,
        filepath: Path,
        mode: str,
        headers: List[str],
        row: Optional[dict],
        max_retries: int = 5,
        base_delay: float = 0.05,
    ) -> None:
        """
        Ouvre un fichier CSV et écrit (header ou ligne) avec retry sur PermissionError.
        L'EA MT5 peut tenir un lock exclusif Windows jusqu'à ~200ms.
        On attend donc 50ms, 100ms, 200ms, 400ms, 800ms avant d'abandonner.
        """
        for attempt in range(max_retries):
            try:
                with open(filepath, mode, newline="", encoding="utf-8") as f:
                    writer = csv.DictWriter(f, fieldnames=headers, extrasaction="ignore")
                    if row is None:
                        writer.writeheader()
                    else:
                        writer.writerow(row)
                return  # succès
            except PermissionError:
                if attempt < max_retries - 1:
                    delay = base_delay * (2 ** attempt)   # 50ms, 100ms, 200ms, 400ms
                    logger.debug(
                        f"[MT5Bridge] PermissionError sur {filepath.name} "
                        f"(tentative {attempt + 1}/{max_retries}) — retry dans {delay*1000:.0f}ms"
                    )
                    time.sleep(delay)
                else:
                    logger.error(
                        f"[MT5Bridge] PermissionError persistant sur {filepath.name} "
                        f"après {max_retries} tentatives — signal perdu."
                    )
                    raise

    class _safe_read:
        """Context manager pour lire un CSV en toute sécurité."""
        def __init__(self, filepath: Path):
            self._filepath = filepath
            self._lock_path = str(filepath) + ".lock"

        def __enter__(self) -> List[dict]:
            if HAS_FILELOCK:
                self._fl = FileLock(self._lock_path, timeout=5)
                self._fl.acquire()
            try:
                # utf-8-sig supprime le BOM (\ufeff) des CSV Windows/MetaTrader 5
                # Sans ca, le 1er header devient \ufeffsignal_id -> KeyError
                with open(self._filepath, "r", newline="", encoding="utf-8-sig") as f:
                    reader = csv.DictReader(f)
                    self._rows = list(reader)
            except (OSError, csv.Error):
                self._rows = []
            return self._rows

        def __exit__(self, *args):
            if HAS_FILELOCK:
                try:
                    self._fl.release()
                except Exception:
                    pass


# ─── Point d'entrée de test ───────────────────────────────────────────────────

if __name__ == "__main__":
    """
    Test rapide du bridge en mode simulation locale.
    Lance ce script pour vérifier que le bridge fonctionne
    avant de le connecter à un vrai EA MT5.

    Usage :
        python mt5_bridge.py
    """
    import tempfile

    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s  %(levelname)-7s  %(message)s",
        datefmt="%H:%M:%S",
    )

    print("\n" + "="*60)
    print("  AHAD QUANT MT5 Bridge — Test de simulation")
    print("="*60 + "\n")

    # Dossier temporaire simulant MQL5/Files/
    with tempfile.TemporaryDirectory() as tmpdir:
        bridge = MT5Bridge(files_path=tmpdir, signal_timeout=10)

        received_reports = []
        received_statuses = []

        bridge.on_report_received(lambda r: received_reports.append(r))
        bridge.on_status_updated(lambda s: received_statuses.append(s))

        bridge.start()

        # ── Test 1 : écriture de signaux ──────────────────────────────
        print("[TEST 1] Écriture de signaux...")
        sig1 = bridge.write_signal("EURUSD", "BUY",  lot=0.10, sl_pips=30, tp_pips=60, confidence=0.82)
        sig2 = bridge.write_signal("GBPUSD", "SELL", lot=0.05, sl_pips=25, tp_pips=50, confidence=0.75)
        sig3 = bridge.write_signal("USDJPY", "BUY",  lot=0.02, sl_pips=40, tp_pips=80, confidence=0.91)
        print(f"  ✅ {sig1.signal_id} | EURUSD BUY")
        print(f"  ✅ {sig2.signal_id} | GBPUSD SELL")
        print(f"  ✅ {sig3.signal_id} | USDJPY BUY")

        # ── Test 2 : simulation réponse EA (rapport OPEN) ─────────────
        print("\n[TEST 2] Simulation rapport OPEN depuis l'EA...")
        report_open = {
            "signal_id":   sig1.signal_id,
            "ticket":      "123456",
            "status":      "OPEN",
            "open_price":  "1.08542",
            "sl":          "1.08242",
            "tp":          "1.09142",
            "open_time":   datetime.now(timezone.utc).isoformat(),
            "close_price": "",
            "close_time":  "",
            "profit":      "",
            "comment":     "Ordre exécuté",
        }
        # Simule l'écriture de l'EA dans reports.csv
        report_path = Path(tmpdir) / "reports.csv"
        with open(report_path, "a", newline="", encoding="utf-8") as f:
            writer = csv.DictWriter(f, fieldnames=MT5Bridge._REPORT_HEADERS)
            writer.writerow(report_open)

        time.sleep(0.3)  # Laisser le watcher détecter
        assert len(received_reports) == 1, f"Attendu 1 rapport, reçu {len(received_reports)}"
        print(f"  ✅ Rapport OPEN reçu : ticket={received_reports[0].ticket}")

        # ── Test 3 : simulation rapport CLOSED ────────────────────────
        print("\n[TEST 3] Simulation rapport CLOSED depuis l'EA...")
        report_closed = {**report_open,
                         "status":      "CLOSED",
                         "close_price": "1.09140",
                         "close_time":  datetime.now(timezone.utc).isoformat(),
                         "profit":      "60.0",
                         "comment":     "TP hit"}
        with open(report_path, "a", newline="", encoding="utf-8") as f:
            writer = csv.DictWriter(f, fieldnames=MT5Bridge._REPORT_HEADERS)
            writer.writerow(report_closed)

        time.sleep(0.3)
        assert len(received_reports) == 2, f"Attendu 2 rapports, reçu {len(received_reports)}"
        print(f"  ✅ Rapport CLOSED reçu : profit={received_reports[1].profit}")

        # ── Test 4 : simulation status EA ─────────────────────────────
        print("\n[TEST 4] Simulation status.csv depuis l'EA...")
        status_path = Path(tmpdir) / "status.csv"
        with open(status_path, "w", newline="", encoding="utf-8") as f:
            writer = csv.DictWriter(f, fieldnames=MT5Bridge._STATUS_HEADERS)
            writer.writeheader()
            writer.writerow({
                "timestamp":      datetime.now(timezone.utc).isoformat(),
                "balance":        "10060.00",
                "equity":         "10045.50",
                "margin":         "120.00",
                "free_margin":    "9925.50",
                "open_positions": "2",
                "daily_pnl":      "60.0",
            })

        time.sleep(0.3)
        status = bridge.read_status()
        assert status is not None
        print(f"  ✅ Statut reçu : balance={status.balance} equity={status.equity}")

        # ── Test 5 : validation des erreurs ───────────────────────────
        print("\n[TEST 5] Validation des paramètres invalides...")
        try:
            bridge.write_signal("EURUSD", "HOLD", lot=0.1, sl_pips=30, tp_pips=60, confidence=0.8)
            print("  ❌ Aurait dû lever une ValueError")
        except ValueError as e:
            print(f"  ✅ ValueError correctement levée : {e}")

        bridge.stop()

        print("\n" + "="*60)
        print("  Tous les tests passés ✅")
        print("="*60 + "\n")