"""
MT5 Execution Layer - Production Direct Mode
Features:
  - Retry with exponential backoff (max 3 attempts, base 0.5s, max 4s)
  - MT5 connection health check + auto-reconnect
  - Order receipt consistency: signal_id ↔ order/deal ID reconciliation
  - Standardized failure types: INIT_FAILED / ORDER_REJECTED / TIMEOUT / PARTIAL_FILL / RETRY_EXHAUSTED
  - Daily rotated statistics logging to logs/mt5_exec_stats.{date}.ndjson
  - intent_id 幂等闭环：每个订单携带唯一 intent_id，执行层去重防止重复下单
"""

import json
import os
import time
import traceback
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, Optional, Tuple

import config_manager

# ─── Constants ────────────────────────────────────────────────────────────────

MAX_RETRIES = 3
BACKOFF_BASE_S = 0.5
BACKOFF_MAX_S = 4.0
CONNECT_TIMEOUT_S = 10
INIT_RETRIES = 2
INIT_BACKOFF_S = 2.0
EA_BRIDGE_TIMEOUT_S = 20.0
EA_BRIDGE_POLL_S = 0.2

_RETRYABLE_CODES = frozenset({
    10027,   # TRADE_RETCODE_REJECT
    10030,   # TRADE_RETCODE_INVALID_PRICES
    10031,   # TRADE_RETCODE_INVALID_VOLUME
    10032,   # TRADE_RETCODE_MARKET_CLOSED
    10033,   # TRADE_RETCODE_TRADE_DISABLED
    10034,   # TRADE_RETCODE_NO_MONEY
    10035,   # TRADE_RETCODE_PRICE_CHANGED
    10036,   # TRADE_RETCODE_OFF_QUOTES
    10038,   # TRADE_RETCODE_INVALID_STOPS
    10039,   # TRADE_RETCODE_TRADE_TOO_LARGE
    10040,   # TRADE_RETCODE_TOO_MANY_REQUESTS
    10041,   # TRADE_RETCODE_HEDGE_DISABLED
    10042,   # TRADE_RETCODE_PROHIBITED_BY_FIFO
})


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()


def _today() -> str:
    return datetime.now(timezone.utc).strftime("%Y%m%d")


# ─── Statistics Logging ──────────────────────────────────────────────────────

def _stats_path(root: Path) -> Path:
    root = root.resolve()
    log_dir = root / "logs"
    log_dir.mkdir(parents=True, exist_ok=True)
    return log_dir / f"mt5_exec_stats.{_today()}.ndjson"


def _log_stat(root: Path, event: Dict[str, Any]) -> None:
    """Append one execution event to daily stats file."""
    try:
        path = _stats_path(root)
        line = json.dumps(event, ensure_ascii=False)
        with path.open("a", encoding="utf-8") as f:
            f.write(line + "\n")
    except Exception:
        pass  # never fail on stats logging


def _load_day_stats(root: Path) -> Dict[str, Any]:
    """Load all stats for today, return aggregated counters."""
    path = _stats_path(root)
    if not path.exists():
        return {}
    totals = {
        "total": 0, "FILLED": 0, "ACKED": 0, "REJECTED": 0,
        "TIMEOUT": 0, "RECONCILED": 0, "PARTIAL_FILL": 0,
        "RETRY_EXHAUSTED": 0, "INIT_FAILED": 0,
        "latencies_ms": [], "retries": 0,
    }
    try:
        for line in path.read_text(encoding="utf-8").splitlines():
            e = json.loads(line)
            for k in totals:
                if k in e:
                    if k == "latencies_ms":
                        totals[k].extend(e[k])
                    elif k == "retries":
                        totals[k] += e[k]
                    else:
                        totals[k] = totals.get(k, 0) + e[k]
    except Exception:
        pass
    return totals


# ─── MT5 Connection Management ───────────────────────────────────────────────

class _MT5Conn:
    """Manages a single MT5 connection with health tracking."""
    __slots__ = ("_mt5", "_last_connect", "_healthy")

    def __init__(self) -> None:
        self._mt5: Any = None
        self._last_connect: float = 0.0
        self._healthy: bool = False

    def _do_init(self) -> bool:
        import MetaTrader5 as mt5
        ok = mt5.initialize()
        if ok:
            self._mt5 = mt5
            self._last_connect = time.time()
            self._healthy = True
        return ok

    def connect(self, force: bool = False) -> bool:
        """Connect if not already healthy. force=True bypasses healthy check."""
        if self._healthy and not force:
            return True
        if self._mt5 is not None:
            try:
                self._mt5.shutdown()
            except Exception:
                pass
            self._mt5 = None
            self._healthy = False
        for attempt in range(INIT_RETRIES):
            if self._do_init():
                return True
            if attempt < INIT_RETRIES - 1:
                time.sleep(INIT_BACKOFF_S * (attempt + 1))
        self._healthy = False
        return False

    @property
    def mt5(self) -> Any:
        return self._mt5

    @property
    def is_healthy(self) -> bool:
        return self._healthy

    def mark_unhealthy(self) -> None:
        self._healthy = False

    def close(self) -> None:
        if self._mt5 is not None:
            try:
                self._mt5.shutdown()
            except Exception:
                pass
            self._mt5 = None
            self._healthy = False


# ─── Global connection pool ───────────────────────────────────────────────────

_conn: Optional[_MT5Conn] = None


def _get_conn() -> _MT5Conn:
    global _conn
    if _conn is None:
        _conn = _MT5Conn()
    return _conn


# ─── intent_id 幂等存储 ─────────────────────────────────────────────────────

import threading
import time as _time


class IntentStore:
    """
    内存中去重存储，防止同一 intent_id 的订单重复发送。
    TTL=300秒（5分钟），自动过期清理。
    """
    __slots__ = ("_pending", "_completed", "_lock", "_ttl")

    def __init__(self, ttl: float = 300.0) -> None:
        self._pending: dict[str, float] = {}   # intent_id → start_time
        self._completed: dict[str, dict] = {}  # intent_id → cached report
        self._lock = threading.Lock()
        self._ttl = ttl

    def try_start(self, intent_id: str) -> Optional[dict]:
        """
        尝试标记一个 intent_id 为进行中。
        返回 None 表示可以继续（未被占用）。
        返回 dict 表示重复，直接返回缓存结果。
        """
        with self._lock:
            self._cleanup_unlocked()
            if intent_id in self._completed:
                return self._completed[intent_id]
            if intent_id in self._pending:
                return None  # 正在处理中，忽略重复
            self._pending[intent_id] = _time.time()
            return None

    def complete(self, intent_id: str, report: dict) -> None:
        """标记 intent_id 完成，并缓存结果"""
        with self._lock:
            self._pending.pop(intent_id, None)
            self._completed[intent_id] = report

    def _cleanup_unlocked(self) -> None:
        """清理过期条目（须在锁内调用）"""
        now = _time.time()
        expired = [k for k, v in self._pending.items() if now - v > self._ttl]
        for k in expired:
            self._pending.pop(k, None)
        expired_c = [k for k, v in self._completed.items() if now - v.get("_completed_at", 0) > self._ttl]
        for k in expired_c:
            self._completed.pop(k, None)


# ─── 全局 intent_id 去重存储 ─────────────────────────────────────────────────
_intent_store: Optional[IntentStore] = None


def _get_intent_store() -> IntentStore:
    global _intent_store
    if _intent_store is None:
        _intent_store = IntentStore(ttl=300.0)
    return _intent_store


# ─── Retry helper ─────────────────────────────────────────────────────────────

def _exponential_backoff(attempt: int) -> float:
    delay = BACKOFF_BASE_S * (2 ** attempt)
    return min(delay, BACKOFF_MAX_S)


# ─── intent_id 生成 ──────────────────────────────────────────────────────────

def generate_intent_id(signal: Dict[str, Any], ts_ms: int = None) -> str:
    """
    生成唯一 intent_id，格式：{signal_id}_{action}_{normalized_price}_{ts_ms}

    intent_id 贯穿 signal → risk → execution 全链路。
    用于执行层去重，防止网络抖动导致重复下单。
    """
    if ts_ms is None:
        ts_ms = int(time.time() * 1000)

    signal_id = str(signal.get("signal_id", "unknown"))
    action = str(signal.get("action", "HOLD")).upper()
    side = str(signal.get("side", "NONE")).upper()

    # 价格标准化（取整数部分，去除小数点）
    entry_raw = signal.get("entry_price", 0) or signal.get("price", 0)
    normalized_price = str(int(entry_raw * 1000))  # 精确到3位小数

    return f"{signal_id}_{action}_{side}_{normalized_price}_{ts_ms}"


def _coerce_float(value: Any, field_name: str, *, allow_zero: bool = True) -> float:
    try:
        number = float(value or 0)
    except (TypeError, ValueError) as exc:
        raise ValueError(f"INVALID_{field_name.upper()}") from exc
    if number < 0 or (not allow_zero and number <= 0):
        raise ValueError(f"INVALID_{field_name.upper()}")
    return number


def _coerce_int(value: Any, field_name: str) -> int:
    try:
        number = int(value or 0)
    except (TypeError, ValueError) as exc:
        raise ValueError(f"INVALID_{field_name.upper()}") from exc
    if number < 0:
        raise ValueError(f"INVALID_{field_name.upper()}")
    return number


def _write_json_atomic(path: Path, payload: Dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temp_path = path.with_suffix(path.suffix + ".tmp")
    temp_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
    temp_path.replace(path)


def _build_ea_bridge_request_payload(
    signal_id: str,
    intent_id: str,
    symbol: str,
    signal: Dict[str, Any],
) -> Dict[str, Any]:
    action = str(signal.get("action", "") or "").upper()
    side = str(signal.get("side", "") or "").upper()
    if not action:
        raise ValueError("EA_REQUEST_ACTION_MISSING")

    volume_raw = signal.get("volume")
    if volume_raw is None:
        volume_raw = signal.get("volume_lots", 0)
    volume = _coerce_float(volume_raw, "volume", allow_zero=False) if action in {"OPEN", "CLOSE_PARTIAL"} else _coerce_float(volume_raw, "volume")
    sl_price = _coerce_float(signal.get("sl_price", 0), "sl_price")
    tp_price = _coerce_float(signal.get("tp_price", 0), "tp_price")
    ticket = _coerce_int(signal.get("ticket", 0), "ticket")

    if action == "OPEN" and side not in {"BUY", "SELL"}:
        raise ValueError("EA_REQUEST_INVALID_SIDE")
    if action in {"CLOSE", "CLOSE_TICKET", "MODIFY_SL", "MODIFY_TP", "CLOSE_PARTIAL"} and not symbol:
        raise ValueError("EA_REQUEST_SYMBOL_MISSING")

    payload = {
        "signal_id": signal_id,
        "intent_id": intent_id,
        "symbol": symbol,
        "action": action,
        "side": side,
        "volume": volume,
        "sl_price": sl_price,
        "tp_price": tp_price,
        "ticket": ticket,
        "signal": signal,
        "ts": _now(),
    }
    if action == "OPEN" and isinstance(signal.get("effective_param_snapshot"), dict):
        payload["effective_param_snapshot"] = signal["effective_param_snapshot"]
    for passthrough_key in ("count", "timeframe", "days", "country"):
        if passthrough_key in signal:
            payload[passthrough_key] = signal[passthrough_key]
    return payload


def _report_ticket(report: Dict[str, Any]) -> int:
    for key in ("ticket", "deal_id", "order_id"):
        value = report.get(key, 0)
        try:
            ticket = int(value or 0)
        except (TypeError, ValueError):
            continue
        if ticket > 0:
            return ticket
    return 0


def _send_order_impl(
    mt5: Any, signal: Dict[str, Any], symbol: str, intent_id: Optional[str] = None
) -> Tuple[bool, int, str, Any]:
    """
    Attempt one order send. Returns (success, retcode, reason).
    intent_id: 用于写入MT5订单comment，便于事后追溯。
    """
    side = str(signal.get("side", "NONE")).upper()
    if side == "BUY":
        order_type = mt5.ORDER_TYPE_BUY
        price = mt5.symbol_info_tick(symbol).ask
    elif side == "SELL":
        order_type = mt5.ORDER_TYPE_SELL
        price = mt5.symbol_info_tick(symbol).bid
    else:
        return False, -1, "INVALID_SIDE", None

    volume = float(signal.get("volume_lots", 0))
    if volume <= 0:
        return False, -2, "INVALID_VOLUME", None

    sl = float(signal.get("sl_price", 0))
    tp = float(signal.get("tp_price", 0))

    request = {
        "action": mt5.TRADE_ACTION_DEAL,
        "symbol": symbol,
        "volume": volume,
        "type": order_type,
        "price": price,
        "sl": sl if sl > 0 else 0,
        "tp": tp if tp > 0 else 0,
        "deviation": 10,
        "type_time": mt5.ORDER_TIME_GTC,
        "type_filling": mt5.ORDER_FILLING_FOK,
        "comment": intent_id or str(signal.get("signal_id", "")),
    }

    result = mt5.order_send(request)
    if result is None:
        return False, -99, "ORDER_SEND_NONE", None

    retcode = int(getattr(result, "retcode", -1))
    if retcode == mt5.TRADE_RETCODE_DONE:
        return True, retcode, "ORDER_FILLED", result

    # Classify failure
    if retcode in _RETRYABLE_CODES:
        return False, retcode, f"MT5_RETCODE_RETRY_{retcode}", result
    return False, retcode, f"MT5_RETCODE_{retcode}", result


# ─── Receipt reconciliation ───────────────────────────────────────────────────

def _reconcile_receipt(
    signal_id: str, result: Any, symbol: str, mt5: Any
) -> Dict[str, Any]:
    """
    Verify deal/order IDs match signal_id and fill in details.
    """
    deal_id = int(getattr(result, "deal", 0))
    order_id = int(getattr(result, "order", 0))
    volume = float(getattr(result, "volume", 0))
    price = float(getattr(result, "price", 0))

    # Cross-check: verify deal exists in MT5
    reconciliation_ok = False
    try:
        if deal_id > 0:
            deals = mt5.history_deals_get(
                datetime.now(timezone.utc).replace(hour=0, minute=0, second=0),
                datetime.now(timezone.utc),
            )
            if deals:
                for d in deals:
                    if int(getattr(d, "ticket", 0)) == deal_id:
                        reconciliation_ok = True
                        break
    except Exception:
        pass

    return {
        "ticket": deal_id or order_id,
        "order_id": order_id,
        "deal_id": deal_id,
        "deal_volume": volume,
        "deal_price": price,
        "reconciliation_ok": reconciliation_ok,
    }


# ─── Main dispatch ────────────────────────────────────────────────────────────

def dispatch_execution(
    signal: Dict[str, Any],
    mode: str,
    symbol: str,
    execution_report_path: Optional[Path],
    intent_id: Optional[str] = None,
) -> Dict[str, Any]:
    """
    Public entry point – matches original function signature.
    Routes to file/mock or production direct mode.
    intent_id: 可选唯一订单标识，用于执行层幂等去重。
    """
    selected_mode = str(mode).lower()
    if selected_mode == "file":
        report = _load_file_report(execution_report_path, str(signal.get("signal_id")))
        if report is not None:
            return report
        return {"signal_id": signal.get("signal_id"), "status": "TIMEOUT",
                "reason": "REPORT_FILE_NOT_READY", "ts": _now()}
    if selected_mode == "mock":
        action = str(signal.get("action", "HOLD"))
        status = "ACKED" if action in {"OPEN", "CLOSE", "MODIFY"} else "RECONCILED"
        return {"signal_id": signal.get("signal_id"), "status": status,
                "reason": "MOCK_EXECUTION", "ts": _now()}
    if selected_mode == "direct":
        return dispatch_mt5_direct(
            signal, symbol,
            _project_root_from_signal(signal),
            intent_id=intent_id,
        )
    if selected_mode == "ea_bridge":
        return dispatch_ea_bridge(
            signal,
            symbol,
            execution_report_path,
            _project_root_from_signal(signal),
            intent_id=intent_id,
        )
    return {"signal_id": signal.get("signal_id"), "status": "REJECTED",
            "reason": "INVALID_EXECUTION_MODE", "ts": _now()}


def dispatch_mt5_direct(
    signal: Dict[str, Any],
    symbol: str,
    project_root: Path,
    intent_id: Optional[str] = None,
) -> Dict[str, Any]:
    """
    Production direct dispatch with retry, reconnect, and reconciliation.
    intent_id: 用于幂等去重，传递 None 则自动生成。
    """
    signal_id = str(signal.get("signal_id", ""))
    # ── intent_id 幂等检查 ──
    if intent_id is None:
        intent_id = generate_intent_id(signal)
    store = _get_intent_store()
    cached = store.try_start(intent_id)
    if cached is not None:
        # 重复请求，直接返回缓存结果（幂等保证）
        return cached
    start_ms = int(time.time() * 1000)

    action = str(signal.get("action", "HOLD")).upper()
    effective_param_snapshot = _ensure_effective_param_snapshot(signal) if action == "OPEN" else None

    # ── Non-order actions ──
    if action == "HOLD":
        result = _report("RECONCILED", signal_id, "HOLD_NO_ORDER", start_ms, project_root)
        result["intent_id"] = intent_id
        store.complete(intent_id, result)
        return result

    if action not in {"OPEN", "CLOSE", "MODIFY"}:
        result = _report("REJECTED", signal_id, "UNSUPPORTED_ACTION", start_ms, project_root)
        result["intent_id"] = intent_id
        store.complete(intent_id, result)
        return result

    # ── Connect to MT5 ──
    conn = _get_conn()
    if not conn.connect():
        conn.mark_unhealthy()
        _log_stat(project_root, {
            "signal_id": signal_id, "status": "INIT_FAILED",
            "reason": "MT5_INIT_FAILED", "ts": _now(), "retries": 0,
        })
        result = _report("TIMEOUT", signal_id, "INIT_FAILED", start_ms, project_root)
        result["intent_id"] = intent_id
        store.complete(intent_id, result)
        return result

    mt5 = conn.mt5
    total_retries = 0

    # ── Retry loop ──
    last_error = "UNKNOWN"
    last_retcode = -1

    for attempt in range(MAX_RETRIES):
        try:
            success, retcode, reason, send_result = _send_order_impl(mt5, signal, symbol, intent_id=intent_id)
            total_retries = attempt

            if success:
                receipt = _reconcile_receipt(signal_id, send_result, symbol, mt5)
                latency_ms = int(time.time() * 1000) - start_ms
                _log_stat(project_root, {
                    "signal_id": signal_id, "status": "FILLED",
                    "reason": "ORDER_FILLED",
                    "retcode": retcode,
                    "order_id": receipt["order_id"],
                    "deal_id": receipt["deal_id"],
                    "reconciliation_ok": receipt["reconciliation_ok"],
                    "latencies_ms": [latency_ms],
                    "retries": total_retries,
                    "ts": _now(),
                })
                result = _report("FILLED", signal_id, "ORDER_FILLED", start_ms, project_root,
                               ticket=receipt["ticket"],
                               order_id=receipt["order_id"], deal_id=receipt["deal_id"],
                               reconciliation_ok=receipt["reconciliation_ok"],
                               latency_ms=latency_ms, retries=total_retries,
                               retcode=retcode,
                               effective_param_snapshot=effective_param_snapshot)
                result["intent_id"] = intent_id
                store.complete(intent_id, result)
                return result

            last_retcode = retcode
            last_error = reason

            # Non-retryable: give up immediately
            if "MT5_RETCODE_RETRY" not in reason:
                break

        except Exception as exc:
            last_error = f"EXCEPTION:{type(exc).__name__}:{exc}"
            last_retcode = -98
            conn.mark_unhealthy()
            # Force reconnect on next attempt
            conn.connect(force=True)

        # Backoff before retry
        if attempt < MAX_RETRIES - 1:
            delay = _exponential_backoff(attempt)
            time.sleep(delay)
            # Attempt reconnect if unhealthy
            if not conn.is_healthy:
                conn.connect(force=True)

    # All retries exhausted
    latency_ms = int(time.time() * 1000) - start_ms

    # Classify final failure
    if "INIT" in last_error or last_retcode in (-99, -98):
        final_status = "TIMEOUT"
        final_reason = "INIT_FAILED" if "INIT" in last_error else last_error
    elif "MT5_RETCODE_RETRY" in last_error:
        final_status = "RETRY_EXHAUSTED"
        final_reason = last_error
    else:
        final_status = "REJECTED"
        final_reason = last_error

    _log_stat(project_root, {
        "signal_id": signal_id, "status": final_status,
        "reason": final_reason,
        "retcode": last_retcode,
        "latencies_ms": [latency_ms],
        "retries": total_retries,
        "ts": _now(),
    })
    result = _report(final_status, signal_id, final_reason, start_ms, project_root,
                   latency_ms=latency_ms, retries=total_retries, retcode=last_retcode,
                   effective_param_snapshot=effective_param_snapshot)
    result["intent_id"] = intent_id
    store.complete(intent_id, result)
    return result


# ─── Helpers ──────────────────────────────────────────────────────────────────

def _project_root_from_signal(signal: Dict[str, Any]) -> Path:
    """Try to derive project root from signal metadata."""
    hint = signal.get("_project_root")
    if hint:
        p = Path(hint)
        if p.exists():
            return p
    # Fallback to this file's location
    return Path(__file__).resolve().parents[1]


def _load_file_report(path: Optional[Path], signal_id: str) -> Optional[Dict[str, Any]]:
    if path is None or not path.exists():
        return None
    try:
        report = json.loads(path.read_text(encoding="utf-8"))
        if report.get("signal_id") != signal_id:
            return None
        return report
    except Exception:
        return None


def _candidate_files_dirs() -> list[Path]:
    home = Path.home()
    prefixes = [
        home / "Library/Application Support/net.metaquotes.wine.metatrader5",
        home / "Library/Application Support/net.metaquotes.wine.metatrader5-beta",
        home / ".wine",
    ]
    out: list[Path] = []
    for prefix in prefixes:
        fixed = [
            prefix / "drive_c/users/user/AppData/Roaming/MetaQuotes/Terminal/Common/Files",
            prefix / f"drive_c/users/{os.getenv('USER', 'user')}/AppData/Roaming/MetaQuotes/Terminal/Common/Files",
            prefix / "drive_c/Program Files/MetaTrader 5/MQL5/Files",
        ]
        for p in fixed:
            if p.exists():
                out.append(p)
        if prefix.exists():
            for p in prefix.glob("**/MetaQuotes/Terminal/*/Common/Files"):
                if p.exists():
                    out.append(p)
    uniq = []
    seen = set()
    for p in out:
        s = str(p.resolve())
        if s in seen:
            continue
        seen.add(s)
        uniq.append(p)
    return uniq


def _wine_mql5_files_dir() -> Optional[Path]:
    """
    Detect Wine MT5 shared Files directory.
    Priority:
      1) Terminal/Common/Files (matches FILE_COMMON used by EA)
      2) MQL5/Files
    """
    env_dir = os.environ.get("EA_BRIDGE_FILES_DIR", "").strip()
    if env_dir:
        p = Path(env_dir).expanduser()
        if p.exists():
            return p
    candidates = _candidate_files_dirs()
    if not candidates:
        return None
    with_snapshot = []
    with_bridge = []
    for files in candidates:
        bridge = files / "gold_autotrade/bridge"
        snap = bridge / "market_snapshot.json"
        if snap.exists():
            with_snapshot.append((snap.stat().st_mtime, files))
        elif bridge.exists():
            with_bridge.append(files)
    if with_snapshot:
        with_snapshot.sort(key=lambda x: x[0], reverse=True)
        return with_snapshot[0][1]
    if with_bridge:
        return with_bridge[0]
    for files in candidates:
        if files.exists():
            return files
    return None


def _ea_bridge_paths(project_root: Path) -> Dict[str, Path]:
    """
    Bridge files go to Wine shared Files directory for EA to read.
    Archive stays in project directory for easy access.
    """
    wine_files = _wine_mql5_files_dir()
    # Use Wine MQL5/Files if available, otherwise fall back to project dir
    if wine_files is not None:
        if (wine_files / "to_ea").exists() or (wine_files / "from_ea").exists():
            root = wine_files
        else:
            root = wine_files / "gold_autotrade/bridge"
        to_ea = root / "to_ea"
        from_ea = root / "from_ea"
        to_ea.mkdir(parents=True, exist_ok=True)
        from_ea.mkdir(parents=True, exist_ok=True)
        # Archive in project dir (easy to inspect)
        archive = (project_root / "data/bridge/archive")
        archive.mkdir(parents=True, exist_ok=True)
    else:
        # Fallback: use project directory (e.g., Linux/Mac no-Wine)
        root = project_root / "data/bridge"
        to_ea = root / "to_ea"
        from_ea = root / "from_ea"
        archive = root / "archive"
        to_ea.mkdir(parents=True, exist_ok=True)
        from_ea.mkdir(parents=True, exist_ok=True)
        archive.mkdir(parents=True, exist_ok=True)
    return {"root": root, "to_ea": to_ea, "from_ea": from_ea, "archive": archive}


def _ensure_effective_param_snapshot(signal: Dict[str, Any]) -> dict[str, Any]:
    existing = signal.get("effective_param_snapshot")
    if isinstance(existing, dict):
        return existing

    snapshot = config_manager.get_effective_params_snapshot()
    snapshot["config_revision"] = str(signal.get("config_revision", "manual") or "manual")
    snapshot["requested_volume_lots"] = float(
        signal.get("requested_volume_lots", signal.get("volume_lots", signal.get("volume", signal.get("lots", 0.0)))) or 0.0
    )
    snapshot["configured_base_lots"] = float(signal.get("configured_base_lots", 0.0) or 0.0)
    snapshot["rule_cap_lots"] = float(signal.get("rule_cap_lots", 0.0) or 0.0)
    snapshot["final_volume_lots"] = float(signal.get("final_volume_lots", signal.get("volume_lots", 0.0)) or 0.0)
    snapshot["rule_cap_reasons"] = (
        list(signal.get("rule_cap_reasons", []))
        if isinstance(signal.get("rule_cap_reasons", []), list)
        else []
    )
    signal["effective_param_snapshot"] = snapshot
    return snapshot


def dispatch_ea_bridge(
    signal: Dict[str, Any],
    symbol: str,
    execution_report_path: Optional[Path],
    project_root: Path,
    intent_id: Optional[str] = None,
) -> Dict[str, Any]:
    signal_id = str(signal.get("signal_id", ""))
    if not signal_id:
        return _report("REJECTED", signal_id, "SIGNAL_ID_MISSING", 0, project_root)

    # ── intent_id 幂等检查 ──
    if intent_id is None:
        intent_id = generate_intent_id(signal)
    store = _get_intent_store()
    cached = store.try_start(intent_id)
    if cached is not None:
        return cached

    start_ms = int(time.time() * 1000)
    action = str(signal.get("action", "HOLD")).upper()
    effective_param_snapshot = _ensure_effective_param_snapshot(signal) if action == "OPEN" else None

    paths = _ea_bridge_paths(project_root)
    try:
        request_payload = _build_ea_bridge_request_payload(signal_id, intent_id, symbol, signal)
    except ValueError as exc:
        result = _report("REJECTED", signal_id, str(exc), start_ms, project_root)
        result["intent_id"] = intent_id
        store.complete(intent_id, result)
        return result
    request_file = paths["to_ea"] / f"{signal_id}.request.json"
    response_file = paths["from_ea"] / f"{signal_id}.report.json"
    _write_json_atomic(request_file, request_payload)
    _write_json_atomic(paths["archive"] / f"{signal_id}.request.json", request_payload)

    timeout_s = EA_BRIDGE_TIMEOUT_S
    if execution_report_path is not None:
        custom_timeout = execution_report_path.with_suffix(".timeout")
        if custom_timeout.exists():
            try:
                timeout_s = float(custom_timeout.read_text(encoding="utf-8").strip())
            except Exception:
                timeout_s = EA_BRIDGE_TIMEOUT_S
    deadline = time.time() + max(1.0, timeout_s)

    while time.time() < deadline:
        if response_file.exists():
            try:
                report = json.loads(response_file.read_text(encoding="utf-8"))
            except Exception:
                report = {"signal_id": signal_id, "status": "TIMEOUT", "reason": "EA_REPORT_INVALID_JSON", "ts": _now()}
            if effective_param_snapshot is not None and not isinstance(report.get("effective_param_snapshot"), dict):
                report["effective_param_snapshot"] = effective_param_snapshot
            report["intent_id"] = intent_id
            ticket = _report_ticket(report)
            if ticket > 0 and not report.get("ticket"):
                report["ticket"] = ticket
            archive_report = paths["archive"] / f"{signal_id}.report.json"
            try:
                response_file.replace(archive_report)
            except Exception:
                pass
            store.complete(intent_id, report)
            return report
        if execution_report_path is not None:
            report = _load_file_report(execution_report_path, signal_id)
            if report is not None:
                if effective_param_snapshot is not None and not isinstance(report.get("effective_param_snapshot"), dict):
                    report["effective_param_snapshot"] = effective_param_snapshot
                report["intent_id"] = intent_id
                store.complete(intent_id, report)
                return report
        time.sleep(EA_BRIDGE_POLL_S)

    _log_stat(project_root, {
        "signal_id": signal_id,
        "status": "TIMEOUT",
        "reason": "EA_BRIDGE_TIMEOUT",
        "ts": _now(),
        "retries": 0,
    })
    result = _report(
        "TIMEOUT",
        signal_id,
        "EA_BRIDGE_TIMEOUT",
        start_ms,
        project_root,
        effective_param_snapshot=effective_param_snapshot,
    )
    result["intent_id"] = intent_id
    store.complete(intent_id, result)
    return result


def _report(
    status: str,
    signal_id: str,
    reason: str,
    start_ms: int,
    project_root: Path,
    **extra: Any,
) -> Dict[str, Any]:
    latency_ms = int(time.time() * 1000) - start_ms
    r = {
        "signal_id": signal_id,
        "status": status,
        "reason": reason,
        "latency_ms": latency_ms,
        "ts": _now(),
    }
    for k, v in extra.items():
        if v is not None:
            r[k] = v
    return r


# ─── Original file/mode dispatch (kept for compatibility) ─────────────────────
