#!/usr/bin/env python3
"""
Gold Trader — 桥接适配层 / 共享工具库 (daemon 已退役, 2026-05-19)

============================================================
!! IMPORTANT: pipeline.py 是唯一交易引擎 !!
!! 本模块仅提供共享工具函数，不做任何交易决策   !!
============================================================

pipeline / gui_backend 安全导入:
  [状态管理]    load_state, save_state, init_db
  [价格源]      get_price, get_price_truth, get_mt5_position_state
  [交易执行]    write_trade_request, reconcile_positions, process_ea_reports
  [持仓管理]    _finalize_position_closure, _apply_realized_closure_stats
  [Agent方向]   get_agent_direction, get_technical_trend_bias, save_agent_direction
  [MT5运维]     is_mt5_running, restart_mt5
  [桥接常量]    _bridge_to_ea_dir(), _bridge_from_ea_dir()
  [日志]        _daemon_log

已退役 (调用会触发 DeprecationWarning):
  daemon_loop, signal_poll_loop, monitor_position_loop, start_daemon,
  become_daemon, _try_pipeline_primary_entry,
  _spawn_macro_direction_process, all signal generation functions
"""

import warnings as _warnings

_DEPRECATED_FUNCTIONS = {
    "daemon_loop", "signal_poll_loop", "monitor_position_loop",
    "start_daemon", "become_daemon", "_try_pipeline_primary_entry",
    "_spawn_macro_direction_process",
}


def __getattr__(name):
    if name in _DEPRECATED_FUNCTIONS:
        _warnings.warn(
            f"bridge.gold_trader.{name} is DEPRECATED. "
            f"Use pipeline.py instead.",
            DeprecationWarning, stacklevel=2,
        )
        raise AttributeError(
            f"bridge.gold_trader.{name} has been retired. Use pipeline.py."
        )
    raise AttributeError(f"module 'bridge.gold_trader' has no attribute '{name}'")
import argparse, json, math, os, re, sys, time, signal, atexit, pathlib, subprocess, sqlite3, uuid, datetime, threading, shlex
from typing import Any, Optional

from src.symbol_resolver import normalize_symbol, pick_best_instrument, symbol_matches_preferred
from src.market_context import (
    is_defensive_regime,
    is_range_regime,
    is_trend_regime,
    normalize_regime_label,
)

# ─── 逼宫机制状态 ───
# 用于记录底层模型连续发起高置信度反向信号的次数和冷却时间
_override_counter = 0
_last_override_signal_time = 0
_last_agent_wakeup_time = 0
_startup_banner_emitted = False
AGENT_WAKEUP_COOLDOWN = 600  # 逼宫唤醒冷却时间（10分钟），保证 M2 失效时能更快重判
_last_macro_poll_time = 0
_last_direction_periodic_refresh = 0
DIRECTION_PERIODIC_REFRESH_SEC = 7200  # 兜底定时刷新：每 2 小时强制检查一次方向
HOLD_MACRO_POLL_MIN_STREAK = 2
HOLD_MACRO_POLL_INTERVAL_SEC = 240

# ── 启动就绪门：防止 config 未加载完时触发 AI 调用 ──
_config_ready = False
_config_ready_checked_at = 0.0

def _validate_config_ready() -> bool:
    """验证核心配置可正常读取。启动时调用，成功后缓存为 True。

    在 config 就绪之前，_spawn_macro_direction_process 拒绝执行，
    避免产生误导性的 "API key not configured" 错误。
    每 30 秒重试一次，确保瞬态故障后自动恢复。
    """
    global _config_ready, _config_ready_checked_at
    if _config_ready:
        return True
    now = time.time()
    if _config_ready_checked_at > 0 and (now - _config_ready_checked_at) < 30:
        return False
    _config_ready_checked_at = now
    try:
        from src.llm.config import load_llm_providers, get_api_key
        providers = load_llm_providers()
        deepseek = providers.get("deepseek", {})
        if not deepseek:
            _daemon_log("[CONFIG_GATE] Provider 'deepseek' not configured — config not ready")
            return False
        key = get_api_key(deepseek)
        if not key:
            _daemon_log("[CONFIG_GATE] DeepSeek API key empty — config not ready")
            return False
        _config_ready = True
        _daemon_log("[CONFIG_GATE] Config validated — AI macro direction available")
        return True
    except Exception as e:
        _daemon_log(f"[CONFIG_GATE] Config check failed: {e}")
        return False
_state_machine = None
from collections import deque
import queue as _queue
_mutiny_evidence_queue: deque = deque(maxlen=5)

_MAX_ZOMBIE_REAP_INTERVAL = 10  # 每 N 秒扫一次僵尸进程

# ============ 全局终止标志（信号处理） ============
KILL_FLAG = threading.Event()

def _sigterm_handler(signum, frame):
    KILL_FLAG.set()
    sys.exit(0)
# 信号处理器注册移至 __main__ --fg 分支，避免作为库被导入时注册

# ============ 路径配置 ============
# 使用 .resolve() 正确解析符号链接，确保无论通过什么路径运行，ROOT 都指向实际文件所在目录
# ROOT now resolved via frozen_path (PyInstaller-safe)
from frozen_path import APP_ROOT as ROOT
sys.path.insert(0, str(ROOT))
import config_manager
from bridge.direction_registry import save_agent_direction as _registry_save_agent_direction
from bridge.direction_registry import load_macro_direction_record as _registry_load_macro_direction_record
from bridge.direction_registry import get_agent_direction as _registry_get_agent_direction
from strategy_runtime import get_strategy_runtime
from src.agent_decision_contract import load_active_interpreter_payload, load_macro_contract_payload, update_macro_contract, DEFAULT_MACRO_TTL_SEC
from src.position_lifecycle import (
    load_position_lifecycle,
    normalize_position_attribution,
    position_state_fields_from_attribution,
    reconcile_position_lifecycle,
    sync_close_fill,
    sync_close_request,
    sync_flat_recovery,
    sync_open_fill,
    sync_open_request,
)
from src.runtime_trade_guard import (
    build_profit_sense_cooldown_profile as _shared_build_profit_sense_cooldown_profile,
    cooldown_remaining_sec as _shared_cooldown_remaining_sec,
    same_side_reentry_veto as _shared_same_side_reentry_veto,
)
from src.strategy_registry import canonical_strategy_id

# 使用 config_manager 动态获取路径
APP_DIR = config_manager.get_app_dir()
DATA_DIR = config_manager.get_data_dir()
LOGS_DIR = config_manager.get_logs_dir()
EXECUTION_STATE_FILE = DATA_DIR / "execution_state.json"

# MT5 通信目录 — 动态获取，支持 GUI 修改后即时生效
def _get_mt5_files_dir():
    return config_manager.get_mt5_dir()

# 初始化 SQLite 状态库
DB_PATH = DATA_DIR / "bridge" / "state" / "agent_state.db"
DB_PATH.parent.mkdir(parents=True, exist_ok=True)


def _open_state_db(*, timeout: float = 10.0, check_same_thread: bool = False) -> sqlite3.Connection:
    # Runtime app dir can drift between dev, launchd, and packaged runs. Recreate
    # the parent on every open so reconciliation does not die on a missing path.
    DB_PATH.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(str(DB_PATH), timeout=timeout, check_same_thread=check_same_thread)
    conn.execute("PRAGMA journal_mode=WAL")
    # 强制同步落盘，防止断电或崩溃时 position_open=False 丢失
    conn.execute("PRAGMA synchronous=FULL")
    conn.execute(f"PRAGMA busy_timeout={int(timeout * 1000)}")
    return conn


def init_db():
    # 使用 check_same_thread=False 解决多进程 fork 时的问题
    conn = _open_state_db(check_same_thread=False)
    c = conn.cursor()
    c.execute("""
    CREATE TABLE IF NOT EXISTS state_kv (
        key TEXT PRIMARY KEY,
        value TEXT,
        updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
    """)
    c.execute("""
    CREATE TABLE IF NOT EXISTS active_positions (
        ticket INTEGER PRIMARY KEY,
        side TEXT,
        volume REAL,
        entry_price REAL,
        sl REAL,
        tp REAL,
        agent_id TEXT,
        opened_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
    """)
    conn.commit()
    conn.close()

# 不要在模块加载时直接初始化，改到主循环里，防止 fork 崩溃
# init_db()

def _bridge_to_ea_dir():
    return _get_mt5_files_dir() / "gold_autotrade/bridge/to_ea"

def _bridge_from_ea_dir():
    return _get_mt5_files_dir() / "gold_autotrade/bridge/from_ea"

# Export aliases for external modules (exit_optimizer.py)
BRIDGE_TO_EA = _bridge_to_ea_dir()
FROM_EA_DIR = _bridge_from_ea_dir()
ARCHIVE_DIR    = DATA_DIR / "bridge/archive"
ARCHIVE_FILE_RETENTION = 1200
LOG_DIR        = LOGS_DIR
STATE_FILE     = DATA_DIR / "bridge/state/agent_state.json"
PID_FILE       = LOGS_DIR / "gold_trader.pid"
DECISION_LINK_FILE = LOGS_DIR / "decision_signal_links.ndjson"
LOCK_FILE      = LOGS_DIR / "gold_trader.lock"
# Repo daemon and launchagent runtime must share one machine-wide singleton
# lock, otherwise legacy watchdogs can create a second execution brain.
def _resolve_global_runtime_dir() -> pathlib.Path:
    if sys.platform == "win32":
        base = pathlib.Path(os.environ.get("APPDATA", pathlib.Path.home() / "AppData" / "Roaming"))
    elif sys.platform == "darwin":
        base = pathlib.Path.home() / "Library" / "Application Support"
    else:
        base = pathlib.Path(os.environ.get("XDG_DATA_HOME", pathlib.Path.home() / ".local" / "share"))
    return base / "LobsterQuant" / "runtime_locks"

GLOBAL_RUNTIME_DIR = _resolve_global_runtime_dir()
GLOBAL_RUNTIME_DIR.mkdir(parents=True, exist_ok=True)
GLOBAL_PID_FILE = GLOBAL_RUNTIME_DIR / "gold_trader.pid"
GLOBAL_LOCK_FILE = GLOBAL_RUNTIME_DIR / "gold_trader.lock"
DAEMON_LOG     = LOGS_DIR / "gold_trader.daemon.log"
DAEMON_LOG.parent.mkdir(parents=True, exist_ok=True)  # 本地目录，避免 iCloud 阻塞
REVIEW_DIR     = LOGS_DIR / "review_pending"
AGENT_DIR      = LOGS_DIR / "agent_decisions"   # Agent 方向决策存储
TRAILING_TELEMETRY_FILE = LOGS_DIR / "trailing_telemetry_latest.json"
TRADE_FEEDBACK_QUEUE_FILE = LOGS_DIR / "trade_feedback_queue.json"

REVIEW_DIR.mkdir(parents=True, exist_ok=True)
STATE_FILE.parent.mkdir(parents=True, exist_ok=True)

INTERVAL_NO_POS  = 120
INTERVAL_HAS_POS = 10
# MIN_CONFIDENCE resolved via config_manager.get_effective_min_confidence_threshold() at runtime
def _get_min_confidence() -> float:
    from config_manager import get_effective_min_confidence_threshold
    return get_effective_min_confidence_threshold()
CONFIDENCE_HARD_FLOOR = 0.62  # 绝对最低置信度，低于此值的开仓信号无条件驳回
# SINGLE_OWNER_RISK_ARBITER unified arbitration mode:
#   When True (default), Governor operates in observe-only mode—risk decisions
#   flow through the single pipeline (risk_manager.py → GoldRiskManager.evaluate).
#   CRITICAL: The following safety nets apply UNCONDITIONALLY regardless of this flag:
#     - Dynamic cooldown after close (防高频追单)
#     - Falling-knife hard block (防接飞刀)
#     - Confidence hard floor (防低质量开仓)
SINGLE_OWNER_RISK_ARBITER = True

# 平仓后冷却硬地板：任何情况下冷却时间不低于此值（秒）
HARD_COOLDOWN_FLOOR_SEC = 5 * 60  # 5 分钟绝对最低冷却
_last_close_time = 0  # 内存镜像，由 state["last_close_time"] 同步更新

# 执行层已知晓信号生成器不产出 volume_lots，此标志仅用于减少重复日志
_volume_lots_warning_logged = False
_last_macro_notice_key = ""
_last_macro_notice_time = 0.0
_last_veto_lifecycle_key = ""
_last_veto_lifecycle_time = 0.0
_last_micro_status_report_time = 0.0
_STALE_WARN_LAST: dict = {}
_STALE_WARN_INTERVAL = 60.0
MONITOR_ONLY = False
_PIPELINE_EA_LIVE_SOURCE = None
PIPELINE_SINGLE_TRACK_AUTOMATION = True
_PIPELINE_BLOCKED_REQUEST_ORIGINS = {
    "legacy_signal_poll",
    "soft_limit_trigger",
    "review_approved",
    "legacy_position_monitor",
    "legacy_macro_escape",
    # legacy_reject_fallback 不拦截：它是 EA 拒绝后的紧急平仓 fallback，
    # 只发 CLOSE/CLOSE_TICKET 不发 OPEN，拦截会造成仓位既平不掉又反复
    # 触发 PARTIAL_CLOSE → 死循环直到 daemon 崩溃。
    "legacy_post_fill_repair",
}

# ============ 日志 ============
def _log(msg, level="INFO"):
    now = time.strftime("%H:%M:%S")
    line = f"[{now}] [{level}] {msg}\n"
    try:
        with open(LOG_DIR / f"gold_trader.{time.strftime('%Y%m%d')}.log", "a", encoding="utf-8") as f:
            f.write(line)
    except Exception:
        pass

def _daemon_log(msg, level="DAEMON"):
    now = time.strftime("%H:%M:%S")
    try:
        with open(DAEMON_LOG, "a", encoding="utf-8") as f:
            tag = str(level or "DAEMON").upper()
            f.write(f"[{now}] [{tag}] {msg}\n")
    except Exception:
        pass


def _prune_archive_dir(directory: pathlib.Path, keep_latest: int = ARCHIVE_FILE_RETENTION) -> None:
    if keep_latest <= 0 or not directory.exists():
        return
    files = [path for path in directory.glob("*.json") if path.is_file()]
    if len(files) <= keep_latest:
        return
    files.sort(key=lambda item: item.stat().st_mtime, reverse=True)
    for target in files[keep_latest:]:
        try:
            target.unlink(missing_ok=True)
        except Exception:
            continue


def _pipeline_single_track_enabled() -> bool:
    return bool(PIPELINE_SINGLE_TRACK_AUTOMATION)


def _resolve_monitor_only_runtime_flag(args: argparse.Namespace) -> bool:
    requested_monitor_only = bool(getattr(args, "monitor_only", False))
    runtime_requested = bool(getattr(args, "daemon", False) or getattr(args, "fg", False))
    if not runtime_requested:
        return requested_monitor_only
    if _pipeline_single_track_enabled():
        if not requested_monitor_only:
            raise SystemExit(
                "PIPELINE_SINGLE_TRACK_REQUIRES_MONITOR_ONLY: "
                "gold_trader cannot run as a trading core; use src/pipeline.py for live trading"
            )
        return True
    return requested_monitor_only


def _enforce_monitor_only_runtime() -> None:
    if _pipeline_single_track_enabled() and not MONITOR_ONLY:
        raise RuntimeError(
            "PIPELINE_SINGLE_TRACK_REQUIRES_MONITOR_ONLY: "
            "gold_trader runtime must stay monitor-only while pipeline owns trading"
        )


def _report_owned_by_pipeline(report: Optional[dict]) -> bool:
    payload = report if isinstance(report, dict) else {}
    if str(payload.get("writer_owner", "") or "").lower() == "pipeline":
        return True
    if str(payload.get("request_origin", "") or "").lower() == "pipeline_dispatch":
        return True
    attribution = payload.get("attribution", {}) if isinstance(payload.get("attribution", {}), dict) else {}
    if str(attribution.get("writer_owner", "") or "").lower() == "pipeline":
        return True
    return False


def _position_owned_by_pipeline(state: dict) -> bool:
    attribution = state.get("position_attribution", {}) if isinstance(state.get("position_attribution", {}), dict) else {}
    if str(attribution.get("writer_owner", "") or "").lower() == "pipeline":
        return True
    risk_decision = attribution.get("risk_decision", {}) if isinstance(attribution.get("risk_decision", {}), dict) else {}
    execution_owner = risk_decision.get("execution_owner", {}) if isinstance(risk_decision.get("execution_owner", {}), dict) else {}
    return str(execution_owner.get("owner_mode", "") or "").lower() == "pipeline_single_track"


def _block_legacy_trade_request(
    action: str,
    request_origin: str,
    *,
    decision_id: str = "",
    side: str = "",
) -> bool:
    action_upper = str(action or "").upper()
    origin = str(request_origin or "bridge_legacy").strip().lower()
    if not _pipeline_single_track_enabled():
        return False
    if action_upper not in {"OPEN", "CLOSE", "REDUCE", "MODIFY_SL", "MODIFY_TP", "CLOSE_ALL", "CLOSE_TICKET"}:
        return False
    if origin not in _PIPELINE_BLOCKED_REQUEST_ORIGINS:
        return False
    reason = f"PIPELINE_SINGLE_TRACK_BLOCK:{origin}:{action_upper}"
    _daemon_log(f"[PIPELINE_OWNER] blocked legacy trade request origin={origin} action={action_upper}")
    if decision_id:
        try:
            _transition_decision_state(
                decision_id,
                "REJECTED",
                reason=reason,
                request_origin=origin,
                request_action=action_upper,
                request_side=str(side or "").upper(),
            )
        except Exception as exc:
            _daemon_log(f"[PIPELINE_OWNER] transition reject failed: {exc}")
    return True


def _build_runtime_trade_summary(state: dict, ledger_path: Optional[pathlib.Path] = None) -> dict:
    summary = {"trades": 0, "pnl": 0.0, "source": "state_kv"}
    target = pathlib.Path(ledger_path) if ledger_path is not None else (ROOT / "logs" / "trading_ledger.ndjson")
    if target.exists():
        trades = 0
        pnl_sum = 0.0
        try:
            with target.open("r", encoding="utf-8") as handle:
                for line in handle:
                    if not line.strip():
                        continue
                    try:
                        row = json.loads(line)
                    except Exception:
                        continue
                    action = str(row.get("action", "CLOSE") or "CLOSE").upper()
                    if action == "OPEN":
                        continue
                    pnl = float(row.get("pnl", 0.0) or 0.0)
                    trades += 1
                    pnl_sum += pnl
            if trades > 0 or abs(pnl_sum) > 0.005:
                return {"trades": trades, "pnl": round(pnl_sum, 2), "source": "trading_ledger"}
        except Exception:
            pass

    rows = state.get("recent_closures", []) if isinstance(state, dict) else []
    if isinstance(rows, list) and rows:
        trades = 0
        pnl_sum = 0.0
        for row in rows:
            if not isinstance(row, dict):
                continue
            try:
                pnl_sum += float(row.get("pnl", 0.0) or 0.0)
                trades += 1
            except Exception:
                continue
        return {"trades": trades, "pnl": round(pnl_sum, 2), "source": "recent_closures"}

    try:
        summary["trades"] = max(0, int((state or {}).get("total_trades", 0) or 0))
    except Exception:
        summary["trades"] = 0
    try:
        summary["pnl"] = round(float((state or {}).get("total_pnl", 0.0) or 0.0), 2)
    except Exception:
        summary["pnl"] = 0.0
    return summary


_RUNTIME_BLACKBOARD = None
_RUNTIME_BLACKBOARD_PATH = APP_DIR / "blackboard_state.json"


def _get_runtime_blackboard():
    global _RUNTIME_BLACKBOARD
    if _RUNTIME_BLACKBOARD is None:
        from src.swarm.blackboard import Blackboard

        # Pin the runtime blackboard to APP_DIR so launchd, release copies, and
        # direct process restarts all converge on the same state file.
        _RUNTIME_BLACKBOARD = Blackboard(str(_RUNTIME_BLACKBOARD_PATH))
    return _RUNTIME_BLACKBOARD


def _runtime_blackboard_update(namespace: str, key: str, value):
    try:
        return _get_runtime_blackboard().update(namespace, key, value)
    except Exception as e:
        _daemon_log(f"[BLACKBOARD_SYNC] update {namespace}.{key} failed: {e}")
        return False


def _runtime_blackboard_log(role: str, msg_type: str, text: str):
    try:
        _get_runtime_blackboard().add_log(role, msg_type, text)
        return True
    except Exception as e:
        _daemon_log(f"[BLACKBOARD_SYNC] add_log {role}.{msg_type} failed: {e}")
        return False


def _refresh_micro_sniper_hold_state(state: dict):
    if not isinstance(state, dict) or not state.get("position_open"):
        return
    market = {}
    micro = {}
    try:
        snapshot = _get_runtime_blackboard().get_snapshot(include_logs=0)
        if isinstance(snapshot, dict):
            market = snapshot.get("Market_Interpreter", {}) if isinstance(snapshot.get("Market_Interpreter", {}), dict) else {}
            micro = snapshot.get("Micro_Sniper", {}) if isinstance(snapshot.get("Micro_Sniper", {}), dict) else {}
    except Exception as e:
        _daemon_log(f"[BLACKBOARD_SYNC] hold snapshot read failed: {e}")

    pos_interp = state.get("position_interpreter", {}) if isinstance(state.get("position_interpreter", {}), dict) else {}
    session_metrics = state.get("position_session_metrics", {}) if isinstance(state.get("position_session_metrics", {}), dict) else {}
    raw_regime = (
        market.get("market_structure")
        or state.get("position_regime")
        or micro.get("current_regime")
        or "UNKNOWN"
    )
    regime = normalize_regime_label(raw_regime)
    counter_trend_risk = str(
        market.get("counter_trend_risk")
        or pos_interp.get("counter_trend_risk")
        or "LOW"
    ).upper()
    execution_advice = str(
        market.get("execution_advice")
        or pos_interp.get("execution_advice")
        or "UNKNOWN"
    ).upper()
    session_bias = str(
        market.get("session_execution_bias")
        or pos_interp.get("session_execution_bias")
        or session_metrics.get("session_mode")
        or ""
    ).upper()

    defense_mode = str(micro.get("defense_mode", "MARKET") or "MARKET").upper()
    try:
        defense_score = float(micro.get("defense_score", 0.0) or 0.0)
    except Exception:
        defense_score = 0.0

    strong_follow_release = (
        is_trend_regime(regime)
        and counter_trend_risk == "LOW"
        and execution_advice == "ALLOW_FOLLOW"
        and session_bias in {"HIGH_LIQUIDITY", "NY_VOLATILE", "LONDON_TREND", "LONDON_VOLATILE", "LONDON_NY_OVERLAP", "OVERLAP_TREND"}
    )

    if strong_follow_release:
        defense_mode = "MARKET"
        defense_score = min(defense_score if defense_score > 0 else 0.22, 0.22)
        is_defensive_env = False
        knife_catching_risk = False
    else:
        if execution_advice == "LIMIT_ONLY" or counter_trend_risk == "HIGH":
            if defense_mode not in {"TIGHT_LIMIT", "LIMIT", "WIDE_LIMIT"}:
                defense_mode = "LIMIT" if counter_trend_risk == "HIGH" else "TIGHT_LIMIT"
            defense_score = max(defense_score, 0.58 if counter_trend_risk == "HIGH" else 0.42)
        is_defensive_env = defense_mode in {"TIGHT_LIMIT", "LIMIT", "WIDE_LIMIT"}
        knife_catching_risk = is_defensive_env and counter_trend_risk == "HIGH"

    previous_regime = normalize_regime_label(micro.get("current_regime", "UNKNOWN"))
    _runtime_blackboard_update("Micro_Sniper", "current_regime", regime)
    _runtime_blackboard_update("Micro_Sniper", "is_defensive_env", bool(is_defensive_env))
    _runtime_blackboard_update("Micro_Sniper", "knife_catching_risk", bool(knife_catching_risk))
    _runtime_blackboard_update("Micro_Sniper", "defense_mode", defense_mode)
    _runtime_blackboard_update("Micro_Sniper", "defense_score", round(defense_score, 4))
    _runtime_blackboard_update("Micro_Sniper", "recent_stop_losses", int(state.get("consecutive_losses", 0) or 0))
    if previous_regime and previous_regime != "UNKNOWN" and previous_regime != regime:
        _runtime_blackboard_log("system", "REGIME_CHANGED", f"{previous_regime}->{regime}")
    # NY时段状态推送到前端
    try:
        from bridge.session_analyzer import get_current_session
        _sym = _detect_active_symbol()
        _cur_sess = str(get_current_session(symbol=_sym)).upper()
        _runtime_blackboard_update("Micro_Sniper", "ny_session_active", _cur_sess in {"NEW_YORK", "LONDON_NY_OVERLAP"})
    except Exception:
        pass


def _read_runtime_drawdown_snapshot() -> tuple[float, str]:
    candidates: list[tuple[str, float]] = []

    try:
        blackboard_file = APP_DIR / "blackboard_state.json"
        if blackboard_file.exists():
            payload = json.loads(blackboard_file.read_text(encoding="utf-8"))
            risk_commissar = payload.get("Risk_Commissar", {}) if isinstance(payload, dict) else {}
            current_drawdown = abs(float(risk_commissar.get("current_drawdown_pct", 0.0) or 0.0))
            if current_drawdown > 0:
                candidates.append(("BLACKBOARD_CURRENT_DRAWDOWN", current_drawdown))
    except Exception:
        pass

    try:
        system_state_file = APP_DIR / "tenants" / "default" / "ops" / "system_state.json"
        if system_state_file.exists():
            payload = json.loads(system_state_file.read_text(encoding="utf-8"))
            sections = payload.get("sections", {}) if isinstance(payload, dict) else {}
            goals = sections.get("goals", {}) if isinstance(sections, dict) else {}
            gates = goals.get("gates", {}) if isinstance(goals, dict) else {}
            dd_gate = gates.get("max_drawdown_pct", {}) if isinstance(gates, dict) else {}
            current_drawdown = abs(float(dd_gate.get("current", 0.0) or 0.0))
            if current_drawdown > 0:
                candidates.append(("OPS_MAX_DRAWDOWN_GATE", current_drawdown))
    except Exception:
        pass

    if not candidates:
        return 0.0, "NONE"

    source, value = max(candidates, key=lambda item: item[1])
    return round(float(value), 4), str(source)


def _get_total_loss_stop_snapshot(state: Optional[dict] = None) -> dict:
    state = state if isinstance(state, dict) else {}
    threshold_pct = max(0.0, float(_get_effective_param("account_total_loss_stop_pct", 30.0) or 0.0))
    candidates: list[tuple[str, float]] = []

    # `account_total_loss_current_pct` is a derived projection field written back
    # into state for the GUI. Feeding it back in here would make any stale value
    # self-perpetuate across reloads and permanently latch the kill switch.
    # Same applies to daily_loss_pct / intraday_drawdown_pct — if equity is zero
    # the persisted value is garbage from a prior bogus calculation.
    equity_sanity = float(state.get("equity", 0.0) or state.get("balance", 0.0) or 0.0)
    for key, source in (
        ("intraday_drawdown_pct", "STATE_INTRADAY_DRAWDOWN"),
        ("daily_loss_pct", "STATE_DAILY_LOSS"),
    ):
        try:
            value = abs(float(state.get(key, 0.0) or 0.0))
            if value > 0 and equity_sanity > 0 and value <= 100.0:
                candidates.append((source, value))
        except Exception:
            pass

    runtime_drawdown_pct, runtime_source = _read_runtime_drawdown_snapshot()
    if runtime_drawdown_pct > 0:
        candidates.append((runtime_source, runtime_drawdown_pct))

    current_source = "NONE"
    current_pct = 0.0
    if candidates:
        current_source, current_pct = max(candidates, key=lambda item: item[1])

    active = threshold_pct > 0 and current_pct >= threshold_pct
    return {
        "threshold_pct": round(float(threshold_pct), 4),
        "current_pct": round(float(current_pct), 4),
        "current_source": str(current_source),
        "active": bool(active),
    }


def _get_trading_state_machine():
    global _state_machine
    if _state_machine is None:
        try:
            from src.state_machine import LedgerStateCallback, TradingStateMachine
            _state_machine = TradingStateMachine(ROOT, state_file=EXECUTION_STATE_FILE)
            _state_machine.register_ledger_callback(LedgerStateCallback(ROOT))
        except Exception as e:
            _daemon_log(f"state machine init failed: {e}")
            _state_machine = False
    return _state_machine or None


def _transition_decision_state(decision_id, to_state, reason=None, **metadata):
    if not decision_id:
        return
    sm = _get_trading_state_machine()
    if sm is None:
        return
    try:
        sm.transition(to_state, trace_id=str(decision_id), reason=reason, **metadata)
    except Exception as e:
        err_text = str(e)
        if to_state == "EXECUTING" and "不能从 EXECUTING 跳转到 EXECUTING" in err_text:
            return
        _daemon_log(f"state transition failed: decision={decision_id} to={to_state} err={e}")


def _close_decision_state(decision_id, to_state, reason=None, **metadata):
    if not decision_id:
        return
    _transition_decision_state(decision_id, to_state, reason=reason, **metadata)
    _transition_decision_state(decision_id, "IDLE", reason="COMPLETE", **metadata)


def _emit_veto_lifecycle(category, reason, cooldown_sec=180, **metadata):
    global _last_veto_lifecycle_key, _last_veto_lifecycle_time
    veto_key = f"{category}:{reason}"
    now_ts = time.time()
    if veto_key == _last_veto_lifecycle_key and (now_ts - _last_veto_lifecycle_time) < cooldown_sec:
        return
    _last_veto_lifecycle_key = veto_key
    _last_veto_lifecycle_time = now_ts
    trace_id = f"{category.lower()}_{int(now_ts)}"

    # Record PnL to persistent ndjson
    try:
        _pnl_path = LOGS_DIR / "trading_pnl.ndjson"
        _pnl_path.parent.mkdir(parents=True, exist_ok=True)
        import json as _j, datetime as _dt
        _p = {
            "ts": _dt.datetime.now().isoformat(),
            "strategy_id": state_before_close.get("position_strategy_id", "") or "",
            "side": state_before_close.get("position_side", "BUY"),
            "pnl": round(close_pnl, 2),
            "entry_price": state_before_close.get("position_entry", 0),
            "exit_price": round(exit_price, 2),
            "duration_min": int(max(0, duration_min or 0)),
            "reason": str(done_reason or "CLOSE"),
            "ticket": state_before_close.get("position_ticket", 0),
        }
        with open(_pnl_path, "a", encoding="utf-8") as _f:
            _f.write(_j.dumps(_p, ensure_ascii=False) + "\n")
        _pos_path = LOGS_DIR.parent / "data" / "position_lifecycle.json"
        if _pos_path.exists():
            _d = _j.loads(_pos_path.read_text(encoding="utf-8"))
            _h = _d.get("history", [])
            if _h:
                _h[-1]["pnl"] = round(close_pnl, 2)
                _h[-1]["close_price"] = round(exit_price, 2)
            _pos_path.write_text(_j.dumps(_d, indent=2, ensure_ascii=False), encoding="utf-8")
    except Exception:
        pass

    _close_decision_state(trace_id, "VETOED", reason=reason, category=category, **metadata)


def _emit_macro_notice(tag, reason, cooldown_sec=180):
    global _last_macro_notice_key, _last_macro_notice_time
    notice_key = f"{tag}:{reason}"
    now = time.time()
    if notice_key == _last_macro_notice_key and (now - _last_macro_notice_time) < cooldown_sec:
        return
    _last_macro_notice_key = notice_key
    _last_macro_notice_time = now
    _log(f"{tag}: {reason}", "WARN")

# ============ 参数建议系统（从 params_advisor.py 读取）============
_PARAMS_CACHE = None
_PARAMS_CACHE_TTL = 300  # 5分钟有效 (可通过 strategy_sync 配置调整)

def _load_params_suggestion(force=False):
    """读取并缓存参数建议，自动应用 HIGH Priority 项"""
    global _PARAMS_CACHE
    now = time.time()
    if not force and _PARAMS_CACHE and (now - _PARAMS_CACHE["_loaded_at"]) < _PARAMS_CACHE_TTL:
        return _PARAMS_CACHE

    try:
        from src.param_flow import load_runtime_param_payload
        data = load_runtime_param_payload()
    except Exception:
        return {}
    if not data:
        return {}

    # 标记加载时间
    data["_loaded_at"] = now
    data["_loaded_from_file"] = str(data.get("_source_file", ""))

    # Governor 熔断器负责实时拦截，参数调整需人工审批（不再自动应用任何参数）

    _PARAMS_CACHE = data
    return data

from config_manager import get_effective_param as _get_effective_param


# ============ GARCH 波动率（与 risk_manager 同源）============
def _get_vol_multiplier() -> dict:
    """获取 GARCH 波动率乘数（复用 garch_volatility 模块）"""
    try:
        sys.path.insert(0, str(pathlib.Path(__file__).parent))
        from garch_volatility import get_vol_multiplier
        return get_vol_multiplier()
    except Exception:
        return {"multiplier": 1.0, "vol_ratio": 1.0, "regime": "UNKNOWN"}

# ============ 交易记忆系统 ============
def _log_open(state, open_ts=None):
    """开仓时记录到记忆系统（trading_ledger.ndjson）"""
    try:
        sys.path.insert(0, str(ROOT))
        from skill_trading_memory import append_trade
        from datetime import datetime, timezone
        attribution = _build_trade_attribution(state=state)
        regime = attribution.get("regime", state.get("position_regime", "UNKNOWN"))
        session_name = (attribution.get("session", {}) or {}).get("session_name", "UNKNOWN")
        if open_ts is not None:
            ts_str = datetime.fromtimestamp(open_ts, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
        else:
            ts_str = datetime.now(timezone.utc).isoformat()
        append_trade(
            ticket=state.get("position_ticket", 0),
            side=state.get("position_side", "BUY"),
            volume=state.get("position_vol", 0.01),
            entry=state.get("position_entry", 0),
            exit_price=0,
            pnl=0,
            duration_min=0,
            session=session_name,
            regime=regime,
            confidence=float(state.get("position_confidence", 0.70) or 0.70),
            strategy=attribution.get("strategy_id", state.get("position_strategy_id", "AUTO")),
            decision_id=attribution.get("decision_id", state.get("position_decision_id", "")),
            signal_id=attribution.get("signal_id", state.get("position_signal_id", "")),
            trace_id=attribution.get("trace_id", state.get("position_trace_id", "")),
            strategy_id=attribution.get("strategy_id", state.get("position_strategy_id", "AUTO")),
            session_metrics=attribution.get("session", {}),
            interpreter=attribution.get("interpreter", {}),
            sl_hit=False,
            tp_hit=False,
            ts=ts_str,
            action="OPEN",
        )
    except Exception as e:
        _daemon_log(f"_log_open failed: {e}")


def _build_profit_sense_snapshot(state, exit_price=0.0):
    try:
        tele = state.get("trailing_telemetry", {}) if isinstance(state.get("trailing_telemetry", {}), dict) else {}
        side = str(state.get("position_side", "") or "").upper()
        entry = float(state.get("position_entry", 0.0) or 0.0)
        peak_profit_atr = float(tele.get("peak_profit_atr", 0.0) or 0.0)
        capture_ratio = float(tele.get("profit_capture_ratio", 0.0) or 0.0)
        if entry > 0 and exit_price > 0 and peak_profit_atr > 0 and capture_ratio > 0:
            favorable_points = (float(exit_price) - entry) if side == "BUY" else (entry - float(exit_price))
            favorable_points = max(0.0, favorable_points)
            atr_est = favorable_points / max(capture_ratio * peak_profit_atr, 1e-6)
            close_profit_atr = favorable_points / max(atr_est, 1e-6) if atr_est > 0 else (capture_ratio * peak_profit_atr)
        else:
            close_profit_atr = max(0.0, capture_ratio * peak_profit_atr)
        close_capture_ratio = (
            max(0.0, min(1.2, close_profit_atr / max(peak_profit_atr, 1e-6)))
            if peak_profit_atr > 0
            else 0.0
        )
        return {
            "trail_phase": str(tele.get("phase", "NONE") or "NONE"),
            "opportunity_state": str(tele.get("opportunity_state", "UNKNOWN") or "UNKNOWN"),
            "peak_profit_atr": round(peak_profit_atr, 4),
            "close_profit_atr": round(float(close_profit_atr or 0.0), 4),
            "profit_capture_ratio": round(float(close_capture_ratio), 4),
            "profit_giveback_atr": round(max(0.0, peak_profit_atr - close_profit_atr), 4),
            "expected_wave_atr": round(float(tele.get("expected_wave_atr", 0.0) or 0.0), 4),
            "wave_completion": round(float(tele.get("wave_completion", 0.0) or 0.0), 4),
            "peak_completion": round(float(tele.get("peak_completion", 0.0) or 0.0), 4),
            "bars_since_peak": int(tele.get("bars_since_peak", 0) or 0),
            "momentum_decay_score": round(float(tele.get("momentum_decay_score", 0.0) or 0.0), 4),
        }
    except Exception:
        return {
            "trail_phase": "NONE",
            "opportunity_state": "UNKNOWN",
            "peak_profit_atr": 0.0,
            "close_profit_atr": 0.0,
            "profit_capture_ratio": 0.0,
            "profit_giveback_atr": 0.0,
            "expected_wave_atr": 0.0,
            "wave_completion": 0.0,
            "peak_completion": 0.0,
            "bars_since_peak": 0,
            "momentum_decay_score": 0.0,
        }


# ============ trigger_reason 映射（前端显示）============

def _resolve_trigger_reason(raw_reason: str) -> str:
    """将内部触发原因映射为前端可显示的 trigger_reason 值。

    Frontend 识别的值: cold_start, hourly, score_jump, ny_session, override
    """
    if not raw_reason:
        return ""
    r = str(raw_reason).upper().strip()
    # 周期刷新 → hourly
    if r in {"PERIODIC_REFRESH", "PIPELINE_PERIODIC_REFRESH", "HOURLY"}:
        return "hourly"
    # 逼宫 / 反向信号 → override
    if r in {"COUNTERTREND_HQ_BLOCKED_STREAK", "OVERRIDE", "MUTINY"}:
        return "override"
    # 冷启动
    if r in {"COLD_START", "STARTUP"}:
        return "cold_start"
    # NY 时段
    if r in {"NY_SESSION", "NY_OPEN", "SESSION_SWITCH"}:
        return "ny_session"
    # 评分剧变
    if r in {"SCORE_JUMP", "VOLATILITY_SURGE"}:
        return "score_jump"
    # 回退
    return ""


def _build_profit_sense_cooldown_profile(state: dict, regime: str = "") -> dict:
    return _shared_build_profit_sense_cooldown_profile(state, regime=regime)


def _seal_close_guard(
    state: dict,
    *,
    closed_side: str = "",
    closed_ticket: int = 0,
    close_action: str = "CLOSE",
    close_ts=None,
) -> None:
    """
    为所有平仓路径统一落盘"刚刚平仓"的护栏状态。
    这样即使 EA 回报顺序抖动，signal poll 也不会立刻重开同向单。
    """
    if not isinstance(state, dict):
        return
    ts_v = float(close_ts or time.time())
    side_u = str(closed_side or state.get("position_side", "") or "").upper()
    action_u = str(close_action or "CLOSE").upper()
    state["last_close_time"] = ts_v
    state["last_closed_ts"] = ts_v
    if side_u:
        state["last_closed_side"] = side_u
    if closed_ticket:
        state["last_closed_ticket"] = int(closed_ticket)
    state["last_close_action"] = action_u

    # Multi-strategy cooldown: track per-strategy close times
    strategy_id = state.get("position_strategy_id", "") or state.get("position_attribution", {}).get("strategy_id", "")
    if strategy_id and side_u:
        strat_times = state.setdefault("_strategy_close_times", {})
        strat_sides = state.setdefault("_strategy_close_sides", {})
        strat_times[strategy_id] = ts_v
        strat_sides[strategy_id] = side_u
        state["last_closed_strategy"] = strategy_id


def _log_trade(state, price_bid, price_ask, close_pnl=0, duration_min=0, close_ts=None):
    """交易完成（平仓）时记录到记忆系统
    
    Args:
        close_ts: EA真实平仓时间（Unix timestamp），如有则传入。
                   EA FILLED报告不含time字段，此时传None，用Python当前时间（事后由sync_mt5_history.py校正）。
    """
    try:
        sys.path.insert(0, str(ROOT))
        from skill_trading_memory import append_trade
        from datetime import datetime, timezone
        pos = state
        sl_hit = close_pnl < 0 and (pos.get("position_side") == "BUY" and price_bid <= pos.get("position_sl") or pos.get("position_side") == "SELL" and price_ask >= pos.get("position_sl"))
        tp_hit = close_pnl > 0 and (pos.get("position_side") == "BUY" and price_ask >= pos.get("position_tp") or pos.get("position_side") == "SELL" and price_bid <= pos.get("position_tp"))
        attribution = _build_trade_attribution(state=pos)
        regime = attribution.get("regime", pos.get("position_regime", "UNKNOWN"))
        session_name = (attribution.get("session", {}) or {}).get("session_name", "UNKNOWN")
        exit_price = (price_bid + price_ask) / 2
        profit_sense = _build_profit_sense_snapshot(pos, exit_price=exit_price)
        # 平仓时间：优先用EA真实时间，否则用Python当前时间（事后由sync_mt5_history.py校正）
        if close_ts is not None:
            ts_str = datetime.fromtimestamp(close_ts, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
        else:
            ts_str = datetime.now(timezone.utc).isoformat()
        # 平仓记录
        append_trade(
            ticket=pos.get("position_ticket", 0),
            side=pos.get("position_side", "BUY"),
            volume=pos.get("position_vol", 0.01),
            entry=pos.get("position_entry", 0),
            exit_price=exit_price,
            pnl=close_pnl,
            duration_min=duration_min,
            session=session_name,
            regime=regime,
            confidence=float(pos.get("position_confidence", 0.70) or 0.70),
            strategy=attribution.get("strategy_id", pos.get("position_strategy_id", "AUTO")),
            decision_id=attribution.get("decision_id", pos.get("position_decision_id", "")),
            signal_id=attribution.get("signal_id", pos.get("position_signal_id", "")),
            trace_id=attribution.get("trace_id", pos.get("position_trace_id", "")),
            strategy_id=attribution.get("strategy_id", pos.get("position_strategy_id", "AUTO")),
            session_metrics=attribution.get("session", {}),
            interpreter=attribution.get("interpreter", {}),
            sl_hit=sl_hit,
            tp_hit=tp_hit,
            trail_phase=profit_sense.get("trail_phase", "NONE"),
            opportunity_state=profit_sense.get("opportunity_state", "UNKNOWN"),
            peak_profit_atr=profit_sense.get("peak_profit_atr", 0.0),
            close_profit_atr=profit_sense.get("close_profit_atr", 0.0),
            profit_capture_ratio=profit_sense.get("profit_capture_ratio", 0.0),
            profit_giveback_atr=profit_sense.get("profit_giveback_atr", 0.0),
            expected_wave_atr=profit_sense.get("expected_wave_atr", 0.0),
            wave_completion=profit_sense.get("wave_completion", 0.0),
            peak_completion=profit_sense.get("peak_completion", 0.0),
            bars_since_peak=profit_sense.get("bars_since_peak", 0),
            momentum_decay_score=profit_sense.get("momentum_decay_score", 0.0),
            ts=ts_str,
        )
    except Exception as e:
        _daemon_log(f"_log_trade failed: {e}")


def _update_shadow_system(position_open=None, total_trades=None, total_pnl=None, win_rate=None):
    """通过统一黑板 API 更新 Shadow_System，避免覆盖多策略命名空间。"""
    try:
        if position_open is not None:
            _runtime_blackboard_update("Shadow_System", "position_open", position_open)
        if total_trades is not None:
            _runtime_blackboard_update("Shadow_System", "total_trades", total_trades)
        if total_pnl is not None:
            _runtime_blackboard_update("Shadow_System", "total_pnl", total_pnl)
        if win_rate is not None:
            _runtime_blackboard_update("Shadow_System", "win_rate", win_rate)
    except Exception as e:
        _daemon_log(f"_update_shadow_system failed: {e}")


def _update_account_state(balance=None, equity=None):
    """通过统一黑板 API 写入账户状态，避免旧 schema 覆盖新黑板。"""
    try:
        if balance is not None:
            _runtime_blackboard_update("Account", "balance", balance)
        if equity is not None:
            _runtime_blackboard_update("Account", "equity", equity)
    except Exception as e:
        _daemon_log(f"_update_account_state failed: {e}")


def _log_risk_event(reason, price_bid, price_ask, spread=0):
    """风控拦截时记录"""
    try:
        sys.path.insert(0, str(ROOT))
        from skill_trading_memory import append_risk_event
        append_risk_event(
            event_type="GOVERNOR_BLOCK",
            reason=reason,
            spread=spread,
            threshold=70,
            action="HOLD",
            price_bid=price_bid,
        )
    except Exception as e:
        _daemon_log(f"_log_risk_event failed: {e}")

def _append_recent_closure(state, pnl, duration_min=0, exit_price=0.0, close_action="CLOSE"):
    """
    维护最近平仓样本，用于 RiskManager 细粒度亏损压力评估。
    """
    try:
        if not isinstance(state, dict):
            return
        pnl_v = float(pnl or 0.0)
        side = str(state.get("position_side", "") or "").upper()
        agent_dir = str(state.get("position_agent_direction", "") or "").upper()
        agent_aligned = (
            (agent_dir == "LONG" and side == "BUY")
            or (agent_dir == "SHORT" and side == "SELL")
        )
        session_metrics = state.get("position_session_metrics", {})
        if not isinstance(session_metrics, dict):
            session_metrics = {}
        interpreter = state.get("position_interpreter", {})
        if not isinstance(interpreter, dict):
            interpreter = {}
        # 策略归属: priority 1=attribution, 2=position_strategy_id, 3=_position_strategy_map
        strategy_id = str(
            (state.get("position_attribution", {}) or {}).get("strategy_id", "")
            or state.get("position_strategy_id", "")
            or ""
        )
        # 兜底: 从 runtime_open_guard.json 的 _position_strategy_map 按 ticket 查找
        if not strategy_id:
            try:
                import json as _json
                from config_manager import get_data_dir as _get_data_dir
                _rog = _get_data_dir() / "runtime_open_guard.json"
                if _rog.exists():
                    _rog_data = _json.loads(_rog.read_text(encoding="utf-8"))
                    _pmap = _rog_data.get("_position_strategy_map", {})
                    if isinstance(_pmap, dict):
                        _ticket = str(state.get("position_ticket", "") or "")
                        if _ticket and _ticket in _pmap:
                            strategy_id = str(_pmap[_ticket] or "")
            except Exception:
                pass
        row = {
            "ts": int(time.time()),
            "pnl": round(pnl_v, 6),
            "entry_confidence": float(state.get("position_confidence", 0.0) or 0.0),
            "duration_min": int(max(0, duration_min or 0)),
            "entry": float(state.get("position_entry", 0.0) or 0.0),
            "exit": float(exit_price or 0.0),
            "side": side,
            "strategy_id": strategy_id,
            "agent_direction": agent_dir,
            "agent_aligned": bool(agent_aligned),
            "session_mode": str(session_metrics.get("session_mode", "") or ""),
            "selection_regime": normalize_regime_label(
                state.get("position_selection_regime", state.get("position_regime", "UNKNOWN"))
            ),
            "regime_source": str(state.get("position_regime_source", "UNKNOWN") or "UNKNOWN").upper(),
            "regime_confidence": float(state.get("position_regime_confidence", 0.0) or 0.0),
            "counter_trend_risk": str(interpreter.get("counter_trend_risk", "LOW") or "LOW").upper(),
            "close_action": str(close_action or "CLOSE"),
            "market_structure": str(interpreter.get("market_structure", "UNKNOWN") or "UNKNOWN").upper(),
            "confirmation_state": str(interpreter.get("confirmation_state", "NONE") or "NONE").upper(),
            "pullback_quality": str(interpreter.get("pullback_quality", "UNKNOWN") or "UNKNOWN").upper(),
            "execution_advice": str(interpreter.get("execution_advice", "UNKNOWN") or "UNKNOWN").upper(),
            "entry_style": str(interpreter.get("entry_style", state.get("position_entry_style", "UNKNOWN")) or "UNKNOWN").upper(),
            "strategy_id": str(state.get("position_entry_owner_strategy_id", state.get("position_strategy_id", "")) or ""),
            "instance_id": str(state.get("position_entry_owner_instance_id", "") or ""),
        }
        row.update(_build_profit_sense_snapshot(state, exit_price=exit_price))
        pattern_payload = None
        try:
            from src.learning.pattern_memory import record_closure_pattern, attach_runtime_adaptation

            pattern_payload = record_closure_pattern(
                state_before_close=state,
                pnl=pnl_v,
                exit_price=exit_price,
                close_action=str(close_action or "CLOSE"),
            )
            row["fingerprint_id"] = str(pattern_payload.get("fingerprint_id", "") or "")
            row["failure_mode"] = str(pattern_payload.get("failure_mode", "NONE") or "NONE")
            row["success_mode"] = str(pattern_payload.get("success_mode", "NONE") or "NONE")
            attach_runtime_adaptation(state, pattern_payload)
        except Exception as e:
            _daemon_log(f"pattern_memory record failed: {e}")
        rows = state.get("recent_closures", [])
        if not isinstance(rows, list):
            rows = []
        rows.append(row)
        state["recent_closures"] = rows[-40:]

        # 记录按策略的平仓时间和方向（多策略独立冷却用）
        if strategy_id:
            strat_times = state.get("_strategy_close_times", {})
            strat_sides = state.get("_strategy_close_sides", {})
            if not isinstance(strat_times, dict): strat_times = {}
            if not isinstance(strat_sides, dict): strat_sides = {}
            strat_times[strategy_id] = int(time.time())
            strat_sides[strategy_id] = side
            state["_strategy_close_times"] = strat_times
            state["_strategy_close_sides"] = strat_sides
    except Exception as e:
        _daemon_log(f"_append_recent_closure failed: {e}")


def _recent_closure_win_rate(state: dict) -> float:
    rows = _recent_closure_rows(state, max_rows=40)
    if not rows:
        return 0.0
    valid = []
    for row in rows:
        try:
            pnl_v = float(row.get("pnl", 0.0) or 0.0)
        except Exception:
            continue
        if abs(pnl_v) < 1e-9:
            continue
        valid.append(row)
    if not valid:
        return 0.0
    wins = sum(1 for row in valid if float(row.get("pnl", 0.0) or 0.0) > 0)
    return wins / max(1, len(valid))


def _is_effective_recent_closure(row: Any) -> bool:
    if not isinstance(row, dict):
        return False
    close_action = str(row.get("close_action", "") or "").upper()
    source = str(row.get("source", "") or "").lower()
    if source == "pipeline_close_detection":
        return False
    if close_action.startswith("PIPELINE_MT5_"):
        return False
    try:
        float(row.get("pnl", 0.0) or 0.0)
    except Exception:
        return False
    return True


def _recent_closure_rows(state: dict, max_rows: int = 40, strategy_id: str = "") -> list[dict]:
    rows = state.get("recent_closures", []) if isinstance(state, dict) else []
    if not isinstance(rows, list):
        return []
    strategy_key = str(strategy_id or "").strip()
    filtered: list[dict] = []
    for row in rows:
        if not _is_effective_recent_closure(row):
            continue
        if strategy_key and str(row.get("strategy_id", "") or "").strip() != strategy_key:
            continue
        filtered.append(row)
    if max_rows > 0:
        return filtered[-max_rows:]
    return filtered


def _recent_win_rate_from_rows(rows: list[dict]) -> tuple[int, float]:
    valid_rows: list[dict] = []
    for row in rows or []:
        if not isinstance(row, dict):
            continue
        try:
            pnl_v = float(row.get("pnl", 0.0) or 0.0)
        except Exception:
            continue
        if abs(pnl_v) < 1e-9:
            continue
        valid_rows.append(row)
    if not valid_rows:
        return 0, 0.0
    wins = sum(1 for row in valid_rows if float(row.get("pnl", 0.0) or 0.0) > 0)
    return len(valid_rows), wins / max(1, len(valid_rows))


def _recent_loss_streak_from_rows(rows: list[dict]) -> int:
    streak = 0
    for row in reversed(rows or []):
        if not isinstance(row, dict):
            continue
        try:
            pnl_v = float(row.get("pnl", 0.0) or 0.0)
        except Exception:
            continue
        if pnl_v < 0:
            streak += 1
            continue
        break
    return streak


def _recalc_consecutive_losses_from_recent(state: dict, max_rows: int = 40) -> int:
    rows = _recent_closure_rows(state, max_rows=max_rows)
    return _recent_loss_streak_from_rows(rows)


def _sync_consecutive_losses_from_recent(state: dict) -> int:
    """单一事实来源：从 recent_closures 计算连亏数并回写 state。

    consecutive_losses 的唯一计算入口。所有路径（平仓回写、自愈修正、
    对账恢复）都必须通过此函数同步，禁止各处自行拼装该字段。
    """
    losses = _recalc_consecutive_losses_from_recent(state)
    state["consecutive_losses"] = losses

    # 同步今日累计亏损占比（用于 blackboard / GUI / metrics）
    try:
        rows = _recent_closure_rows(state, max_rows=100)
        if rows:
            today_start = int(time.time() // 86400 * 86400)
            today_pnl = sum(
                float(r.get("pnl", 0.0) or 0.0)
                for r in rows
                if int(r.get("ts", 0) or 0) >= today_start
            )
            equity = float(state.get("balance", 0.0) or state.get("equity", 0.0) or 0.0)
            if today_pnl < 0 and equity > 0:
                state["daily_loss_pct"] = round(abs(today_pnl) / equity * 100.0, 4)
            else:
                state["daily_loss_pct"] = 0.0
    except Exception:
        state.setdefault("daily_loss_pct", 0.0)

    return losses


def _apply_realized_closure_stats(state: dict, realized_pnl: float, *, position_open: bool) -> None:
    if not isinstance(state, dict):
        return
    try:
        pnl_v = float(realized_pnl or 0.0)
    except Exception:
        pnl_v = 0.0
    _sync_consecutive_losses_from_recent(state)
    state["total_pnl"] = float(state.get("total_pnl", 0.0) or 0.0) + pnl_v
    state["win_rate"] = _recent_closure_win_rate(state)
    _update_shadow_system(
        position_open=position_open,
        total_trades=state.get("total_trades", 0),
        total_pnl=state.get("total_pnl", 0.0),
        win_rate=state.get("win_rate", 0.0),
    )


def _recent_win_rate_from_state(state: dict, max_rows: int = 20) -> tuple[int, float]:
    rows = _recent_closure_rows(state, max_rows=max_rows)
    return _recent_win_rate_from_rows(rows)


def _same_side_reentry_veto(state: dict, side: str, strategy_id: str = "") -> tuple[int, str]:
    vol_info = _get_vol_multiplier()
    vol_ratio = float(vol_info.get("vol_ratio", 1.0) or 1.0)
    return _shared_same_side_reentry_veto(state, side, vol_ratio=vol_ratio, strategy_id=strategy_id)


def _self_heal_consecutive_losses(state: dict) -> bool:
    if not isinstance(state, dict):
        return False
    old_v = int(state.get("consecutive_losses", 0) or 0)
    new_v = _sync_consecutive_losses_from_recent(state)
    if old_v == new_v:
        return False
    state["win_rate"] = _recent_closure_win_rate(state)
    _runtime_blackboard_update("Micro_Sniper", "recent_stop_losses", new_v)
    _update_shadow_system(
        position_open=bool(state.get("position_open", False)),
        total_trades=state.get("total_trades", 0),
        total_pnl=state.get("total_pnl", 0.0),
        win_rate=state.get("win_rate", 0.0),
    )
    save_state(state)
    _daemon_log(
        f"[STATE_HEAL_CONSEC_LOSSES] corrected consecutive_losses: {old_v} -> {new_v} "
        f"(recent_closures={len(state.get('recent_closures', [])) if isinstance(state.get('recent_closures', []), list) else 0})"
    )
    return True


def _update_defense_metrics(state: dict, mode: str, reason_codes: list[str], score: float) -> bool:
    if not isinstance(state, dict):
        return False
    metrics = state.get("defense_metrics", {})
    if not isinstance(metrics, dict):
        metrics = {}
    mode_u = str(mode or "MARKET").upper()
    mode_counts = metrics.get("mode_counts", {})
    reason_counts = metrics.get("reason_counts", {})
    if not isinstance(mode_counts, dict):
        mode_counts = {}
    if not isinstance(reason_counts, dict):
        reason_counts = {}
    total = int(metrics.get("total", 0) or 0) + 1
    mode_counts[mode_u] = int(mode_counts.get(mode_u, 0) or 0) + 1
    for code in reason_codes[:6]:
        code_u = str(code or "UNKNOWN")
        reason_counts[code_u] = int(reason_counts.get(code_u, 0) or 0) + 1
    metrics.update(
        {
            "total": total,
            "mode_counts": mode_counts,
            "reason_counts": reason_counts,
            "last_mode": mode_u,
            "last_reason_codes": reason_codes[:6],
            "last_score": round(float(score or 0.0), 4),
            "updated_at": int(time.time()),
        }
    )
    state["defense_metrics"] = metrics
    # 周期性持久化，兼顾可观测和性能。
    return (total % 5 == 0) or (mode_u in {"LIMIT", "WIDE_LIMIT"})


def _count_delta(curr: dict, base: dict) -> dict:
    out = {}
    curr = curr if isinstance(curr, dict) else {}
    base = base if isinstance(base, dict) else {}
    keys = set(curr.keys()) | set(base.keys())
    for k in keys:
        out[k] = max(0, int(curr.get(k, 0) or 0) - int(base.get(k, 0) or 0))
    return out


def _persist_trailing_telemetry(telemetry: dict) -> None:
    if not isinstance(telemetry, dict) or not telemetry:
        return
    payload = dict(telemetry)
    payload.setdefault("updated_at", int(time.time()))
    try:
        # 原子写入：先写临时文件再替换，防止读取半截文件
        tmp_path = TRAILING_TELEMETRY_FILE.with_suffix(".tmp")
        tmp_path.write_text(
            json.dumps(payload, ensure_ascii=False, indent=2),
            encoding="utf-8")
        os.replace(tmp_path, TRAILING_TELEMETRY_FILE)
    except Exception as e:
        _daemon_log(f"[TRAIL_TELEMETRY] persist failed: {e}")


def _load_latest_trailing_telemetry() -> dict:
    try:
        if not TRAILING_TELEMETRY_FILE.exists():
            return {}
        payload = json.loads(TRAILING_TELEMETRY_FILE.read_text(encoding="utf-8"))
        return payload if isinstance(payload, dict) else {}
    except json.JSONDecodeError:
        # 文件写入中途被读取 → 尝试从 .tmp 恢复
        tmp_path = TRAILING_TELEMETRY_FILE.with_suffix(".tmp")
        if tmp_path.exists():
            try:
                payload = json.loads(tmp_path.read_text(encoding="utf-8"))
                return payload if isinstance(payload, dict) else {}
            except Exception:
                pass
        _daemon_log("[TRAILING] failed to load telemetry — file corrupted, returning empty")
        return {}
    except Exception as e:
        _daemon_log(f"[TRAILING] failed to load telemetry: {e}")
        return {}


def _emit_defense_matrix_daily_report(state: dict, report_date=None) -> bool:
    if not isinstance(state, dict):
        return False
    metrics = state.get("defense_metrics", {})
    if not isinstance(metrics, dict) or int(metrics.get("total", 0) or 0) <= 0:
        return False
    day_str = report_date or datetime.datetime.now().strftime("%Y-%m-%d")
    if str(state.get("defense_report_last_date", "")) == day_str:
        return False

    baseline = state.get("defense_report_baseline", {})
    if not isinstance(baseline, dict):
        baseline = {}
    curr_total = int(metrics.get("total", 0) or 0)
    base_total = int(baseline.get("total", 0) or 0)
    mode_counts = metrics.get("mode_counts", {}) if isinstance(metrics.get("mode_counts", {}), dict) else {}
    reason_counts = metrics.get("reason_counts", {}) if isinstance(metrics.get("reason_counts", {}), dict) else {}
    mode_delta = _count_delta(mode_counts, baseline.get("mode_counts", {}))
    reason_delta = _count_delta(reason_counts, baseline.get("reason_counts", {}))
    total_delta = max(0, curr_total - base_total)

    top_reasons = sorted(reason_delta.items(), key=lambda x: x[1], reverse=True)[:4]
    trailing = state.get("trailing_telemetry", {})
    trailing_snapshot = {}
    if isinstance(trailing, dict) and trailing:
        trailing_snapshot = {
            "phase": str(trailing.get("phase", "NONE") or "NONE"),
            "trail_basis": float(trailing.get("trail_basis", 0.0) or 0.0),
            "trail_dist_atr": float(trailing.get("trail_dist_atr", 0.0) or 0.0),
            "blocked_count": int(trailing.get("sl_move_step_blocked_count", 0) or 0),
            "opportunity_state": str(trailing.get("opportunity_state", "UNKNOWN") or "UNKNOWN"),
            "capture_ratio": float(trailing.get("profit_capture_ratio", 0.0) or 0.0),
            "wave_completion": float(trailing.get("wave_completion", 0.0) or 0.0),
            "giveback_atr": float(trailing.get("profit_giveback_atr", 0.0) or 0.0),
            "updated_at": int(trailing.get("updated_at", 0) or 0),
        }
    payload = {
        "date": day_str,
        "generated_at": datetime.datetime.now().isoformat(timespec="seconds"),
        "total_delta": total_delta,
        "mode_delta": mode_delta,
        "reason_delta_top4": top_reasons,
        "last_mode": str(metrics.get("last_mode", "UNKNOWN") or "UNKNOWN"),
        "last_score": float(metrics.get("last_score", 0.0) or 0.0),
        "trailing_snapshot": trailing_snapshot,
    }

    report_file = LOGS_DIR / "defense_matrix_daily.json"
    history = []
    try:
        if report_file.exists():
            old = json.loads(report_file.read_text(encoding="utf-8"))
            if isinstance(old, list):
                history = [x for x in old if isinstance(x, dict) and str(x.get("date", "")) != day_str]
    except Exception:
        history = []
    history.append(payload)
    history = history[-45:]
    try:
        report_file.write_text(json.dumps(history, ensure_ascii=False, indent=2), encoding="utf-8")
    except Exception as e:
        _daemon_log(f"[DEFENSE_DAILY] write failed: {e}")

    try:
        m = mode_delta
        reason_txt = ", ".join([f"{k}:{v}" for k, v in top_reasons]) if top_reasons else "no_reason_delta"
        trail_txt = (
            f" trail={trailing_snapshot.get('phase', 'NONE')}"
            f"/basis={float(trailing_snapshot.get('trail_basis', 0.0) or 0.0):.2f}"
            f"/dist={float(trailing_snapshot.get('trail_dist_atr', 0.0) or 0.0):.2f}"
            f"/blocked={int(trailing_snapshot.get('blocked_count', 0) or 0)}"
            if trailing_snapshot
            else " trail=none"
        )
        msg = (
            f"[Defense日报 {day_str}] total={total_delta} "
            f"market={int(m.get('MARKET', 0))} tight={int(m.get('TIGHT_LIMIT', 0))} "
            f"limit={int(m.get('LIMIT', 0))} wide={int(m.get('WIDE_LIMIT', 0))} "
            f"top_reasons={reason_txt}{trail_txt}"
        )
        _runtime_blackboard_log("Risk_Commissar", "REPORT", msg)
    except Exception as e:
        _daemon_log(f"[DEFENSE_DAILY] webchat push failed: {e}")

    state["defense_report_last_date"] = day_str
    state["defense_report_baseline"] = {
        "total": curr_total,
        "mode_counts": mode_counts,
        "reason_counts": reason_counts,
    }
    save_state(state)
    _daemon_log(f"[DEFENSE_DAILY] published for {day_str}: total_delta={total_delta}")
    return True


def _record_learning_observation(decision_id, state, bid, ask, spread, price_source, price_age, agent_direction, signal):
    try:
        from src.learning.decision_recorder import record_observation, record_proposal
        now_ts = time.time()
        indicators = signal.get("indicators", {}) or {}
        attribution = _build_trade_attribution(signal=signal, state=state, agent_direction=agent_direction)
        structure_metrics = _extract_structure_metrics(attribution.get("interpreter", {}))
        payload = {
            "market": {
                "bid": bid,
                "ask": ask,
                "spread": spread,
                "price_source": price_source,
                "price_age": price_age,
                "regime": signal.get("regime", "UNKNOWN"),
                "indicators": indicators,
                "session": attribution.get("session", {}),
                "interpreter": attribution.get("interpreter", {}),
                "structure": structure_metrics,
            },
            "position": {
                "position_open": state.get("position_open", False),
                "position_side": state.get("position_side"),
                "position_entry": state.get("position_entry", 0),
                "position_vol": state.get("position_vol", 0),
                "position_ticket": state.get("position_ticket", 0),
                "time_since_close": max(0, now_ts - float(state.get("last_close_time", 0.0) or 0.0)),
            },
            "risk": {
                "daily_loss_pct": state.get("daily_loss_pct", 0.0),
                "intraday_drawdown_pct": state.get("intraday_drawdown_pct", 0.0),
                "consecutive_losses": state.get("consecutive_losses", 0),
                "kill_switch": state.get("kill_switch", False),
            },
            "macro": {
                "agent_direction": agent_direction,
            },
            "structure": structure_metrics,
            "attribution": attribution,
        }
        proposal = {
            "action": signal.get("action", "HOLD"),
            "side": signal.get("side"),
            "confidence": signal.get("confidence", 0),
            "entry": signal.get("entry", 0),
            "sl": signal.get("sl", 0),
            "tp": signal.get("tp", 0),
            "volume_lots": signal.get("volume_lots", 0),
            "reason": signal.get("reason", ""),
            "regime": normalize_regime_label(signal.get("regime", "UNKNOWN")),
            "strategy_id": signal.get("_sig_strategy_id"),
            "signal_trace_id": signal.get("_sig_trace_id"),
            "selection_regime": normalize_regime_label(
                signal.get("_selection_regime", signal.get("selection_regime", signal.get("regime", "UNKNOWN")))
            ),
            "regime_source": str(signal.get("regime_source", "UNKNOWN") or "UNKNOWN").upper(),
            "regime_confidence": float(signal.get("regime_confidence", 0.0) or 0.0),
            "selected_strategy": signal.get("_selected_strategy"),
            "session": attribution.get("session", {}),
            "interpreter": attribution.get("interpreter", {}),
            "structure": structure_metrics,
            "attribution": attribution,
        }
        record_observation(decision_id, "bridge.gold_trader", payload)
        record_proposal(decision_id, "bridge.gold_trader", proposal)
    except Exception as e:
        _daemon_log(f"_record_learning_observation failed: {e}")


def _extract_session_metrics(session_info):
    if not isinstance(session_info, dict):
        return {}
    analysis = session_info.get("analysis", {}) if isinstance(session_info.get("analysis", {}), dict) else {}
    current = analysis.get("current_analysis", {}) if isinstance(analysis.get("current_analysis", {}), dict) else {}
    session_profile = session_info.get("session_profile", {}) if isinstance(session_info.get("session_profile", {}), dict) else {}
    metrics = {
        "session_name": session_info.get("session_name", analysis.get("current_session", "UNKNOWN")),
        "session_label": session_info.get("session_label", analysis.get("current_session_name", "未知时段")),
        "session_mode": session_info.get("session_mode", session_profile.get("mode", "NORMAL")),
        "recommendation": session_info.get("recommendation", "CAUTIOUS"),
        "reason": session_info.get("reason", ""),
        "direction": current.get("direction"),
        "ema_bias": current.get("ema_bias"),
        "score": current.get("score"),
        "adx": current.get("adx"),
        "atr_pct": current.get("atr_pct"),
        "rsi": current.get("rsi"),
        "trend_strength": current.get("trend_strength"),
        "entry_style": session_info.get("entry_style", session_profile.get("entry_style", "MARKET")),
        "risk_multiplier": session_info.get("risk_multiplier", session_profile.get("risk_multiplier", 1.0)),
    }
    return {k: v for k, v in metrics.items() if v is not None}


def _format_session_metrics(session_metrics):
    if not isinstance(session_metrics, dict) or not session_metrics:
        return ""
    parts = []
    for key in ("session_label", "session_mode", "direction", "ema_bias", "score", "adx", "atr_pct", "rsi", "trend_strength"):
        value = session_metrics.get(key)
        if value in (None, ""):
            continue
        parts.append(f"{key}={value}")
    return f" [{' | '.join(parts)}]" if parts else ""


def _build_interpreter_context(override_data=None, agent_multiplier=1.0, reason="", applied=False):
    if not isinstance(override_data, dict) and not reason and agent_multiplier == 1.0:
        return {}
    override_data = override_data if isinstance(override_data, dict) else {}
    payload = {
        "view": override_data.get("view"),
        "buy_multiplier": override_data.get("buy_multiplier"),
        "sell_multiplier": override_data.get("sell_multiplier"),
        "applied_multiplier": agent_multiplier,
        "reason": reason or override_data.get("reason", ""),
        "valid_until": override_data.get("valid_until"),
        "source": override_data.get("source", "market_interpreter"),
        "applied": applied,
        "market_structure": override_data.get("market_structure"),
        "dominant_direction": override_data.get("dominant_direction"),
        "shock_direction": override_data.get("shock_direction"),
        "pullback_phase": override_data.get("pullback_phase"),
        "pullback_depth_bucket": override_data.get("pullback_depth_bucket"),
        "pullback_quality": override_data.get("pullback_quality"),
        "confirmation_state": override_data.get("confirmation_state"),
        "retrace_ratio": override_data.get("retrace_ratio"),
        "session_name": override_data.get("session_name"),
        "session_execution_bias": override_data.get("session_execution_bias"),
        "structure_signature": override_data.get("structure_signature"),
        "liquidity_sweep": override_data.get("liquidity_sweep"),
        "failed_breakout": override_data.get("failed_breakout"),
        "open_drive": override_data.get("open_drive"),
        "gap_continuation": override_data.get("gap_continuation"),
        "impulse_exhaustion": override_data.get("impulse_exhaustion"),
        "counter_trend_risk": override_data.get("counter_trend_risk"),
        "execution_advice": override_data.get("execution_advice"),
    }
    return {k: v for k, v in payload.items() if v not in (None, "", {})}

def _extract_structure_metrics(interpreter):
    interpreter = interpreter if isinstance(interpreter, dict) else {}
    metrics = {
        "market_structure": interpreter.get("market_structure", "UNCERTAIN"),
        "dominant_direction": interpreter.get("dominant_direction", "MIXED"),
        "shock_direction": interpreter.get("shock_direction", "NONE"),
        "pullback_phase": interpreter.get("pullback_phase", "NONE"),
        "pullback_depth_bucket": interpreter.get("pullback_depth_bucket", "NONE"),
        "pullback_quality": interpreter.get("pullback_quality", "UNKNOWN"),
        "confirmation_state": interpreter.get("confirmation_state", "NONE"),
        "retrace_ratio": interpreter.get("retrace_ratio", 0.0),
        "session_name": interpreter.get("session_name", "UNKNOWN"),
        "session_execution_bias": interpreter.get("session_execution_bias", "BALANCED"),
        "counter_trend_risk": interpreter.get("counter_trend_risk", "LOW"),
        "execution_advice": interpreter.get("execution_advice", "ALLOW_FOLLOW"),
        "liquidity_sweep": interpreter.get("liquidity_sweep", "NONE"),
        "failed_breakout": interpreter.get("failed_breakout", "NONE"),
        "open_drive": bool(interpreter.get("open_drive", False)),
        "gap_continuation": bool(interpreter.get("gap_continuation", False)),
        "impulse_exhaustion": bool(interpreter.get("impulse_exhaustion", False)),
        "view": interpreter.get("view"),
        "source": interpreter.get("source"),
        "applied_multiplier": interpreter.get("applied_multiplier"),
    }
    structure_signature = "|".join(
        str(metrics.get(key, ""))
        for key in ("market_structure", "shock_direction", "pullback_phase", "pullback_depth_bucket", "confirmation_state", "session_name")
    )
    metrics["structure_signature"] = interpreter.get("structure_signature", structure_signature)
    return metrics


def _build_trade_attribution(signal=None, state=None, agent_direction=None, override_data=None, agent_multiplier=1.0, interpreter_reason="", interpreter_applied=False):
    signal = signal if isinstance(signal, dict) else {}
    state = state if isinstance(state, dict) else {}
    existing = signal.get("attribution", {}) if isinstance(signal.get("attribution", {}), dict) else {}
    state_existing = state.get("position_attribution", {}) if isinstance(state.get("position_attribution", {}), dict) else {}
    session_info = signal.get("session") if isinstance(signal.get("session"), dict) else state.get("position_session_info", {})
    session_metrics = signal.get("session_metrics") if isinstance(signal.get("session_metrics"), dict) else state.get("position_session_metrics", {})
    if not session_metrics:
        session_metrics = existing.get("session", {}) or state_existing.get("session", {})
    if not session_metrics:
        session_metrics = _extract_session_metrics(session_info)
    interpreter = signal.get("interpreter", {}) if isinstance(signal.get("interpreter", {}), dict) else state.get("position_interpreter", {})
    if not interpreter:
        interpreter = _build_interpreter_context(
            override_data=override_data,
            agent_multiplier=agent_multiplier,
            reason=interpreter_reason,
            applied=interpreter_applied,
        )
    attribution = {
        "decision_id": signal.get("_learning_decision_id") or signal.get("signal_id") or state.get("position_decision_id") or existing.get("decision_id") or state_existing.get("decision_id"),
        "signal_id": signal.get("signal_id") or state.get("position_signal_id") or existing.get("signal_id") or state_existing.get("signal_id"),
        "trace_id": signal.get("_sig_trace_id") or state.get("position_trace_id") or existing.get("trace_id") or state_existing.get("trace_id"),
        "strategy_id": (
            signal.get("attributed_strategy_id")
            or signal.get("entry_owner_strategy_id")
            or signal.get("_sig_strategy_id")
            or state.get("position_entry_owner_strategy_id")
            or state.get("position_strategy_id")
            or existing.get("strategy_id")
            or state_existing.get("strategy_id")
        ),
        "regime": normalize_regime_label(
            signal.get("selection_regime")
            or signal.get("regime")
            or state.get("position_selection_regime")
            or state.get("position_regime")
            or state.get("regime")
            or existing.get("selection_regime")
            or existing.get("regime")
            or state_existing.get("selection_regime")
            or state_existing.get("regime")
            or "UNKNOWN"
        ),
        "selection_regime": normalize_regime_label(
            signal.get("selection_regime")
            or state.get("position_selection_regime")
            or state.get("position_regime")
            or existing.get("selection_regime")
            or existing.get("regime")
            or state_existing.get("selection_regime")
            or state_existing.get("regime")
            or "UNKNOWN"
        ),
        "regime_source": signal.get("regime_source") or state.get("position_regime_source") or existing.get("regime_source") or state_existing.get("regime_source") or "UNKNOWN",
        "regime_confidence": signal.get("regime_confidence") if signal.get("regime_confidence") is not None else state.get("position_regime_confidence") or existing.get("regime_confidence") or state_existing.get("regime_confidence"),
        "confidence": signal.get("confidence") if signal.get("confidence") is not None else state.get("position_confidence") or existing.get("confidence") or state_existing.get("confidence"),
        "session": session_metrics or existing.get("session", {}) or state_existing.get("session", {}),
        "interpreter": interpreter or existing.get("interpreter", {}) or state_existing.get("interpreter", {}),
        "agent_direction": agent_direction or signal.get("agent_direction") or state.get("position_agent_direction") or existing.get("agent_direction") or state_existing.get("agent_direction"),
    }
    normalized = {
        k: v
        for k, v in attribution.items()
        if v not in (None, "") and not (isinstance(v, dict) and not v)
    }
    return normalize_position_attribution(normalized, action=str(signal.get("action", "") or ""), fallback=state_existing)


def _record_learning_risk_verdict(decision_id, signal, risk_decision):
    try:
        from src.learning.decision_recorder import record_risk_verdict
        risk_payload = risk_decision.get("payload", {}) if isinstance(risk_decision, dict) else {}
        session_info = risk_payload.get("session_info", {})
        session_metrics = _extract_session_metrics(session_info)
        structure_metrics = _extract_structure_metrics(risk_payload.get("interpreter", signal.get("interpreter", {})))
        payload = {
            "proposal_action": signal.get("action", "HOLD"),
            "proposal_side": signal.get("side"),
            "proposal_confidence": signal.get("confidence", 0),
            "final_action": risk_payload.get("action", "HOLD"),
            "final_side": risk_payload.get("side"),
            "drop_reason": risk_payload.get("drop_reason"),
            "lots": risk_payload.get("lots", 0),
            "sl_price": risk_payload.get("sl_price", 0),
            "tp_price": risk_payload.get("tp_price", 0),
            "risk_checks": risk_payload.get("risk_checks", []),
            "session_info": session_info,
            "session_metrics": session_metrics,
            "interpreter": risk_payload.get("interpreter", signal.get("interpreter", {})),
            "structure": structure_metrics,
        }
        record_risk_verdict(decision_id, "bridge.gold_trader", payload)
    except Exception as e:
        _daemon_log(f"_record_learning_risk_verdict failed: {e}")


def _record_learning_execution(decision_id, report, state):
    try:
        from src.learning.decision_recorder import record_execution
        attribution = _build_trade_attribution(state=state)
        structure_metrics = _extract_structure_metrics(attribution.get("interpreter", {}))
        payload = {
            "signal_id": report.get("signal_id", ""),
            "status": report.get("status", ""),
            "action": report.get("action", ""),
            "side": report.get("side", state.get("position_side")),
            "ticket": report.get("ticket", 0) or state.get("position_ticket", 0),
            "fill_price": report.get("fill_price", 0) or state.get("position_entry", 0),
            "volume": report.get("volume", 0.0) or state.get("position_vol", 0.0),
            "sl_price": report.get("sl_price", 0) or state.get("position_sl", 0),
            "tp_price": report.get("tp_price", 0) or state.get("position_tp", 0),
            "regime": attribution.get("regime", "UNKNOWN"),
            "strategy_id": attribution.get("strategy_id"),
            "session": attribution.get("session", {}),
            "interpreter": attribution.get("interpreter", {}),
            "structure": structure_metrics,
            "attribution": attribution,
        }
        record_execution(decision_id, "bridge.gold_trader", payload)
    except Exception as e:
        _daemon_log(f"_record_learning_execution failed: {e}")


def _record_learning_closure(decision_id, state_before_close, pnl, fill_price, duration_min, spread=0.0):
    try:
        from src.learning.decision_recorder import record_closure
        from src.learning.pattern_memory import build_closure_fingerprint, classify_failure_success
        attribution = _build_trade_attribution(state=state_before_close)
        structure_metrics = _extract_structure_metrics(attribution.get("interpreter", {}))
        fingerprint_id, fingerprint = build_closure_fingerprint(state_before_close)
        failure_mode, success_mode = classify_failure_success(
            state_before_close,
            pnl,
            fill_price,
            "CLOSE",
        )
        next_state = {
            "position_open": False,
            "position_ticket": 0,
            "position_side": None,
            "position_entry": 0.0,
            "position_vol": 0.0,
            "position_time": 0,       # 清除旧时间戳，防止 exit_optimizer 误用残留值触发假 TIME_STOP
            "last_close_time": time.time(),
        }
        payload = {
            "ticket": state_before_close.get("position_ticket", 0),
            "side": state_before_close.get("position_side"),
            "entry": state_before_close.get("position_entry", 0),
            "exit_price": fill_price,
            "realized_pnl": pnl,
            "duration_min": duration_min,
            "spread": spread,
            "daily_loss_pct": state_before_close.get("daily_loss_pct", 0.0),
            "consecutive_losses": state_before_close.get("consecutive_losses", 0),
            "done_reason": "trade_closed",
            "next_state": next_state,
            "risk_rejected": False,
            "regime": attribution.get("regime", "UNKNOWN"),
            "strategy_id": attribution.get("strategy_id"),
            "session": attribution.get("session", {}),
            "interpreter": attribution.get("interpreter", {}),
            "structure": structure_metrics,
            "attribution": attribution,
            "fingerprint_id": fingerprint_id,
            "fingerprint": fingerprint,
            "failure_mode": failure_mode,
            "success_mode": success_mode,
        }
        record_closure(decision_id, "bridge.gold_trader", payload)
    except Exception as e:
        _daemon_log(f"_record_learning_closure failed: {e}")


def _record_trade_feedback(state_before_close, close_pnl, exit_price, duration_min, done_reason, decision_id):
    """Record trade closure event for tactical feedback (Phase 2)."""
    def _read_feedback_queue() -> list[dict]:
        try:
            if not TRADE_FEEDBACK_QUEUE_FILE.exists():
                return []
            payload = json.loads(TRADE_FEEDBACK_QUEUE_FILE.read_text(encoding="utf-8"))
            if isinstance(payload, list):
                return [item for item in payload if isinstance(item, dict)]
        except Exception:
            pass
        return []

    def _write_feedback_queue(items: list[dict]) -> None:
        _write_json_atomic(TRADE_FEEDBACK_QUEUE_FILE, items)

    def _enqueue_feedback_retry(
        feedback_data: dict[str, Any],
        state_snapshot: dict[str, Any],
        agent_contract: dict[str, Any],
        features: dict[str, Any] | None,
        error: Exception,
    ) -> None:
        queue = _read_feedback_queue()
        trace_id = str(feedback_data.get("trace_id", "") or "")
        filtered = [item for item in queue if str(item.get("trace_id", "") or "") != trace_id]
        retry_count = 1
        for item in queue:
            if str(item.get("trace_id", "") or "") == trace_id:
                retry_count = int(item.get("retry_count", 0) or 0) + 1
                break
        filtered.append(
            {
                "trace_id": trace_id,
                "queued_at": int(time.time()),
                "retry_count": retry_count,
                "last_error": repr(error),
                "feedback_data": feedback_data,
                "state_before_close": state_snapshot,
                "agent_contract": agent_contract if isinstance(agent_contract, dict) else {},
                "features": features if isinstance(features, dict) else {},
            }
        )
        _write_feedback_queue(filtered)

    def _persist_feedback_features(features: dict[str, Any], queue_on_fail: bool = False) -> bool:
        from src.learning.trading_brain import get_brain

        entry_features = features.get("entry_features", {}) if isinstance(features, dict) else {}
        outcome = features.get("outcome", {}) if isinstance(features, dict) else {}
        brain = get_brain()
        insert_result = brain.record_trade(entry_features, outcome)
        if insert_result is None:
            raise RuntimeError(
                "trade_events insert returned None "
                f"decision_id={entry_features.get('decision_id', '')} "
                f"signal_id={entry_features.get('signal_id', '')}"
            )
        try:
            entry_price = float(entry_features.get("entry_price", 0.0) or 0.0)
            exit_price = float(outcome.get("exit_price", 0.0) or 0.0)
            agent_dir = str(entry_features.get("agent_direction", "") or "")
            price_drift_pct = 0.0
            if entry_price > 0 and exit_price > 0:
                raw_drift = (exit_price - entry_price) / entry_price * 100.0
                price_drift_pct = round(raw_drift, 4)

            brain.auto_validate_latest_judgment(
                trade_side=str(entry_features.get("side", "") or ""),
                pnl=float(outcome.get("pnl", 0.0) or 0.0),
                agent_direction=agent_dir,
                price_drift_pct=price_drift_pct,
                aligned_win_rate=None,
                counter_win_rate=None,
            )
        except Exception:
            pass
        try:
            from src.learning.knowledge_writer import write_trade_note

            write_trade_note(entry_features, outcome)
        except Exception:
            pass
        return True

    def _flush_feedback_queue() -> int:
        queue = _read_feedback_queue()
        if not queue:
            return 0
        remaining: list[dict] = []
        flushed = 0
        for item in queue:
            try:
                features = item.get("features", {})
                if not isinstance(features, dict) or not isinstance(features.get("entry_features", {}), dict):
                    from src.learning.feature_extractor import extract_trade_features

                    features = extract_trade_features(
                        item.get("feedback_data", {}) if isinstance(item.get("feedback_data", {}), dict) else {},
                        item.get("state_before_close", {}) if isinstance(item.get("state_before_close", {}), dict) else {},
                        item.get("agent_contract", {}) if isinstance(item.get("agent_contract", {}), dict) else {},
                    )
                _persist_feedback_features(features, queue_on_fail=False)
                flushed += 1
            except Exception as queue_err:
                item["retry_count"] = int(item.get("retry_count", 0) or 0) + 1
                item["last_error"] = repr(queue_err)
                remaining.append(item)
        if remaining:
            _write_feedback_queue(remaining)
        elif TRADE_FEEDBACK_QUEUE_FILE.exists():
            try:
                TRADE_FEEDBACK_QUEUE_FILE.unlink()
            except Exception:
                pass
        return flushed

    try:
        from src.system_bus import append_trade_event
        flushed = _flush_feedback_queue()
        if flushed > 0:
            _daemon_log(f"[FEEDBACK] flushed pending feedback items={flushed}")
        side = str(state_before_close.get("position_side", "") or "").upper()
        agent_dir = str(state_before_close.get("position_agent_direction", "") or "").upper()
        interpreter = state_before_close.get("position_interpreter", {}) if isinstance(state_before_close.get("position_interpreter", {}), dict) else {}
        feedback_data = {
            "trace_id": decision_id or state_before_close.get("position_decision_id", "") or f"unknown_{time.time_ns()}",
            "decision_id": decision_id or state_before_close.get("position_decision_id", "") or "",
            "signal_id": str(state_before_close.get("position_signal_id", "") or ""),
            "ticket": int(state_before_close.get("position_ticket", 0) or 0),
            "side": side or "UNKNOWN",
            "pnl": round(float(close_pnl or 0), 2),
            "confidence": float(state_before_close.get("position_confidence", 0) or 0),
            "exit_price": float(exit_price or 0),
            "holding_sec": float(duration_min or 0) * 60,
            "agent_direction": agent_dir or "UNKNOWN",
            "regime": str(state_before_close.get("position_regime", "UNKNOWN") or "UNKNOWN"),
            "entry_price": float(state_before_close.get("position_entry", 0) or 0),
            "close_reason": str(done_reason or "UNKNOWN"),
            "strategy_id": str(state_before_close.get("position_strategy_id", "")
                               or state_before_close.get("position_entry_owner_strategy_id", "")),
            "session": str(state_before_close.get("position_session", "") or ""),
            "volume_lots": float(state_before_close.get("position_vol", 0) or 0),
            "adx": float(interpreter.get("adx", 0) or 0),
            "atr": float(interpreter.get("atr", interpreter.get("atr_value", 0)) or 0),
            "spread": float(interpreter.get("spread", 0) or 0),
        }
        try:
            append_trade_event(ROOT, "default", feedback_data)
        except Exception as bus_err:
            _daemon_log(f"[FEEDBACK] append_trade_event failed: {bus_err}")
        # ── Record to learning brain ──
        agent_contract = {}
        try:
            from src.learning.feature_extractor import extract_trade_features
            try:
                from src.agent_decision_contract import load_agent_decision_contract
                agent_contract = load_agent_decision_contract() or {}
            except Exception:
                pass
            features = extract_trade_features(feedback_data, state_before_close, agent_contract)
            _persist_feedback_features(features, queue_on_fail=False)
        except Exception as e:
            _daemon_log(f"[FEEDBACK] record_trade_feedback failed: {e}")
            try:
                _enqueue_feedback_retry(
                    feedback_data=feedback_data,
                    state_snapshot=state_before_close if isinstance(state_before_close, dict) else {},
                    agent_contract=agent_contract,
                    features=features if "features" in locals() and isinstance(features, dict) else {},
                    error=e,
                )
            except Exception as queue_err:
                _daemon_log(f"[FEEDBACK] queue_trade_feedback failed: {queue_err}")
        _daemon_log(f"[FEEDBACK] Trade recorded: {feedback_data['trace_id'][:16]} pnl={feedback_data['pnl']}")
        # Log to strategy_feedback.ndjson for the contextual bandit
        try:
            from src.strategy_feedback import log_feedback
            _sid = str(feedback_data.get("strategy_id", "") or "")
            _regime = str(feedback_data.get("regime", "UNKNOWN") or "UNKNOWN")
            _session = str(feedback_data.get("session", "") or "")
            _adx = float(feedback_data.get("adx", 0) or 0)
            _pnl = float(feedback_data.get("pnl", 0) or 0)
            _is_win = _pnl > 0
            _raw_w = float(feedback_data.get("confidence", 0.5) or 0.5)
            _side = str(feedback_data.get("side", "") or "")
            log_feedback(
                ROOT,
                strategy_id=_sid,
                regime=_regime,
                session=_session,
                adx=_adx,
                vol_ratio=1.0,
                atr_expansion=1.0,
                raw_weight=_raw_w,
                corrected_weight=_raw_w,
                pnl=_pnl,
                is_win=_is_win,
                side=_side,
            )
        except Exception:
            pass

        # Meta decision threshold calibration — backtest each threshold on this trade
        try:
            from src.learning.trading_brain import get_brain
            from src.learning.meta_decision import get_threshold as _gd, record_outcome as _ro
            from bridge.commander_client import _wilson_bound

            _dir = str(feedback_data.get("agent_direction", "") or "").upper()
            if _dir in ("LONG", "SHORT"):
                brain = get_brain()
                _stats = brain.query_composite_win_rate(
                    {"regime": _regime, "session": _session, "agent_direction": _dir},
                    min_samples=5,
                )
                _n = _stats.get("n", 0)
                if _n >= 5:
                    _wr = _stats.get("win_rate", 0.5)
                    _net_pnl = _stats.get("net_pnl", 0.0)
                    _pf = _stats.get("profit_factor", 0.0)
                    _wins = int(round(_n * _wr))

                    _veto_thr = _gd("veto_win_rate")
                    _penalty_thr = _gd("penalty_win_rate")
                    _boost_thr = _gd("boost_win_rate")
                    _cons_wr = _wilson_bound(_wins, _n, z=_gd("num_veto_se"))
                    _opt_wr = _wilson_bound(_wins, _n, z=_gd("num_boost_se"), upper=True)

                    _would_veto = _cons_wr < _veto_thr and _net_pnl < 0 and _pf < 1.0
                    _would_penalize = (not _would_veto) and _cons_wr < _veto_thr + 0.10
                    _would_boost = _opt_wr >= _boost_thr and _pf >= 1.2

                    _veto_correct = _would_veto and (not _is_win)
                    _veto_wrong = _would_veto and _is_win
                    if _would_veto:
                        _ro("veto_win_rate", was_applied=True, was_correct=(not _is_win))
                    else:
                        _ro("veto_win_rate", was_applied=False, was_correct=_is_win)

                    if _would_penalize:
                        _ro("penalty_win_rate", was_applied=True, was_correct=(not _is_win))
                    else:
                        _ro("penalty_win_rate", was_applied=False, was_correct=_is_win)

                    if _would_boost:
                        _ro("boost_win_rate", was_applied=True, was_correct=_is_win)
                    else:
                        _ro("boost_win_rate", was_applied=False, was_correct=(not _is_win))
        except Exception:
            pass
    except Exception as e:
        _daemon_log(f"[FEEDBACK] record_trade_feedback failed: {e}")


# Dedup cache: prevent double-finalization for same ticket
_CLOSED_TICKET_CACHE: set = set()

def _finalize_position_closure(state_before_close, exit_price, close_pnl, duration_min, done_reason, close_ts=None, signal_id_override="", recovered=False):
    ticket = int(state_before_close.get("position_ticket", 0) or 0)
    if ticket > 0 and ticket in _CLOSED_TICKET_CACHE:
        return
    if ticket > 0:
        _CLOSED_TICKET_CACHE.add(ticket)
    decision_id = str(
        state_before_close.get("position_decision_id", "")
        or _resolve_decision_id(
            signal_id=signal_id_override or state_before_close.get("position_signal_id", ""),
            ticket=state_before_close.get("position_ticket", 0),
        )
        or ""
    ).strip()
    if not decision_id:
        decision_id = f"ema_legacy_{state_before_close.get('position_ticket', 'unknown')}"
    _append_recent_closure(
        state_before_close,
        close_pnl,
        duration_min=duration_min,
        exit_price=exit_price,
        close_action=str(done_reason or "CLOSE"),
    )
    _append_decision_link(
        signal_id=signal_id_override or state_before_close.get("position_signal_id", "") or decision_id,
        decision_id=decision_id,
        action="CLOSE",
        ticket=state_before_close.get("position_ticket", 0),
        source=done_reason.lower(),
    )
    _log_trade(
        state_before_close,
        exit_price,
        exit_price,
        close_pnl=close_pnl,
        duration_min=duration_min,
        close_ts=close_ts,
    )
    _record_learning_closure(
        decision_id,
        state_before_close,
        close_pnl,
        exit_price,
        duration_min,
        spread=0.0,
    )
    try:
        from src.live_performance_factors import refresh_live_performance_factors
        refresh_live_performance_factors()
    except Exception as e:
        _daemon_log(f"refresh_live_performance_factors failed: {e}")
    _close_decision_state(
        decision_id,
        "FILLED",
        reason=done_reason,
        ticket=state_before_close.get("position_ticket", 0),
        signal_id=signal_id_override or state_before_close.get("position_signal_id", ""),
        side=state_before_close.get("position_side"),
        exit_price=exit_price,
        realized_pnl=close_pnl,
        recovered=recovered,
    )
    # Phase 2: Record tactical feedback for closed trade
    _record_trade_feedback(state_before_close, close_pnl, exit_price, duration_min, done_reason, decision_id)
    # Ring 2/3 learning cycle trigger (non-blocking)
    try:
        from src.learning.brain_scheduler import run_learning_cycles
        run_learning_cycles()
    except Exception:
        pass

    # Phase 1: shadow close log for counterfactual analysis
    try:
        _sd_entry = {
            "event": "shadow_close",
            "shadow_id": int(time.time() * 1000000) % 10**12,
            "ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "close_ts": close_ts or time.time(),
            "selected_id": str(state_before_close.get("position_strategy_id", "")),
            "side": str(state_before_close.get("position_side", "")),
            "entry_price": float(state_before_close.get("position_entry", 0) or 0),
            "exit_price": float(exit_price or 0),
            "selected_pnl": round(float(close_pnl or 0), 2),
            "duration_min": int(duration_min or 0),
            "exit_reason": str(done_reason or ""),
            "fingerprint": {
                "regime": str(state_before_close.get("position_regime", "")),
                "session": str(state_before_close.get("position_session", "")),
            },
        }
        _sc_path = ROOT / "logs" / "shadow_closes.ndjson"
        _sc_path.parent.mkdir(parents=True, exist_ok=True)
        with open(_sc_path, "a", encoding="utf-8") as _sc_f:
            _sc_f.write(json.dumps(_sd_entry, ensure_ascii=False) + chr(10))
    except Exception:
        pass

    return decision_id

# ============ MT5 ============
def is_mt5_running():
    """Check if MT5 bridge data is available (delegates to Data Hub)."""
    try:
        from src.mt5_data_hub import hub
        return hub.is_connected()
    except Exception:
        pass
    if sys.platform == "win32":
        try:
            output = subprocess.check_output(
                "tasklist /NH /FI \"IMAGENAME eq terminal64.exe\"",
                shell=True, timeout=5, text=True, stderr=subprocess.DEVNULL,
            )
            if "terminal64.exe" in output:
                return True
            # 品牌 MT5 可能用不同进程名（WeTrade, FTMO 等）
            output2 = subprocess.check_output(
                "tasklist /NH /FO CSV",
                shell=True, timeout=5, text=True, stderr=subprocess.DEVNULL,
            )
            for line in output2.splitlines():
                if any(kw in line.lower() for kw in ("metatrader", "wetrade", "ftmo", "terminal64")):
                    return True
            return False
        except Exception:
            return False
    return bool(os.popen("ps aux | grep -i 'metatrader\\|metaquotes\\|wine64.*mt5' | grep -v grep").read().strip())

def restart_mt5(force_kill=False):
    try:
        if force_kill and is_mt5_running():
            subprocess.run(
                "pkill -f 'MetaTrader 5|metaquotes|wine64.*mt5'",
                shell=True,
                capture_output=True,
                text=True,
                timeout=8,
            )
            time.sleep(2)

        app_cfg = config_manager.load_app_config()
        launch_cmd = str(app_cfg.get("mt5_launch_command", "")).strip()
        if launch_cmd:
            cmd_list = shlex.split(launch_cmd)
            subprocess.Popen(
                cmd_list,
                cwd=str(ROOT),
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
            )
        elif sys.platform == "darwin":
            subprocess.Popen(
                ["open", "-a", "MetaTrader 5"],
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
            )
        else:
            return False

        time.sleep(8)
        return is_mt5_running()
    except Exception as e:
        _daemon_log(f"restart_mt5 failed: {e}")
        return False

_last_mt5_error_time = 0


def _refresh_mt5_bridge_dirs():
    # bridge paths are now dynamic — call _bridge_to_ea_dir() / _bridge_from_ea_dir()
    return _get_mt5_files_dir()


def _resolve_mt5_terminal_status_file():
    mt5_files = _refresh_mt5_bridge_dirs()
    candidates = []
    try:
        from src.ea_bridge_paths import resolve_bridge_layout
        layout = resolve_bridge_layout(ROOT)
        candidates.extend([
            layout["bridge_root"] / "from_ea" / "terminal_status.json",
            layout["root"] / "terminal_status.json"
        ])
    except Exception as e:
        _log(f"Failed to resolve bridge layout: {e}", "WARN")

    candidates.extend([
        mt5_files / "gold_autotrade/bridge/from_ea/terminal_status.json",
        mt5_files / "gold_autotrade/bridge/terminal_status.json",
    ])
    # Also check Common terminal (FILE_COMMON writes there)
    _common = pathlib.Path(os.environ.get("APPDATA", "")) / "MetaQuotes" / "Terminal" / "Common" / "Files"
    candidates.extend([
        _common / "gold_autotrade/bridge/from_ea/terminal_status.json",
        _common / "gold_autotrade/bridge/terminal_status.json",
    ])
    for candidate in candidates:
        if candidate.exists():
            return candidate
    return None  # No terminal_status.json found
def _empty_terminal_quote_result(
    *,
    preferred_symbol: str = "XAUUSD",
    source_path: Optional[pathlib.Path] = None,
    reason: str = "MISSING",
) -> dict[str, Any]:
    normalized_preferred = str(preferred_symbol or "XAUUSD").strip().upper() or "XAUUSD"
    return {
        "ok": False,
        "marketSymbol": normalized_preferred,
        "bid": None,
        "ask": None,
        "spread": None,
        "quoteOrigin": "invalid",
        "quoteSymbol": "",
        "topLevelSymbol": "",
        "sourcePath": str(source_path) if source_path is not None else "",
        "schemaState": "invalid",
        "reason": reason,
        "ageSeconds": None,
        "source": "error",
    }


def _extract_terminal_quote(
    data: dict[str, Any],
    *,
    preferred_symbol: str = "XAUUSD",
) -> tuple[Optional[float], Optional[float], Optional[float], str]:
    instruments = data.get("instruments", {}) if isinstance(data.get("instruments", {}), dict) else {}
    normalized_preferred = normalize_symbol(preferred_symbol)

    if instruments:
        matched_symbol, payload = pick_best_instrument(instruments, normalized_preferred)
        if matched_symbol and isinstance(payload, dict):
            try:
                bid = float(payload.get("bid", 0) or 0)
                ask = float(payload.get("ask", 0) or 0)
                spread = float(payload.get("spread", 0) or 0)
            except Exception:
                bid = ask = spread = 0.0
            if bid > 0 and ask > 0:
                return bid, ask, spread, f"instrument:{matched_symbol}"

    try:
        bid = float(data.get("bid", 0) or 0)
        ask = float(data.get("ask", 0) or 0)
        spread = float(data.get("spread", 0) or 0)
    except Exception:
        return None, None, None, "invalid"
    if bid > 0 and ask > 0:
        root_symbol = normalize_symbol(data.get("symbol", ""), default="")
        origin = (
            "root"
            if symbol_matches_preferred(root_symbol, normalized_preferred)
            else f"root:{root_symbol or 'UNKNOWN'}"
        )
        return bid, ask, spread, origin
    return None, None, None, "missing"


_DETECTED_SYMBOL_CACHE: str = ""
_DETECTED_SYMBOL_TS: float = 0.0


def _detect_active_symbol() -> str:
    """Auto-detect the active trading symbol from EA bridge data or MT5 chart.

    Priority: MT5 chart symbol > market_snapshot.json > XAUUSD fallback.
    Cached for 30 seconds to avoid repeated file reads.
    """
    global _DETECTED_SYMBOL_CACHE, _DETECTED_SYMBOL_TS
    now = time.time()
    if _DETECTED_SYMBOL_CACHE and (now - _DETECTED_SYMBOL_TS) < 30:
        return _DETECTED_SYMBOL_CACHE

    # 1. Try MT5 chart symbol (EA is attached to this chart)
    try:
        import MetaTrader5 as _mt5
        if _mt5.initialize():
            chart_id = _mt5.ChartID()
            if chart_id:
                sym = _mt5.ChartSymbol(chart_id)
                if sym:
                    _DETECTED_SYMBOL_CACHE = str(sym).strip().upper()
                    _DETECTED_SYMBOL_TS = now
                    return _DETECTED_SYMBOL_CACHE
    except Exception:
        pass

    # 2. Try terminal_status.json via _resolve_mt5_terminal_status_file (finds correct Common/Files path)
    try:
        snap = _resolve_mt5_terminal_status_file()
        if snap and snap.exists():
            data = json.loads(snap.read_text(encoding="utf-8"))
            sym = data.get("symbol", "") or data.get("marketSymbol", "")
            if sym:
                _DETECTED_SYMBOL_CACHE = str(sym).strip().upper()
                _DETECTED_SYMBOL_TS = now
                return _DETECTED_SYMBOL_CACHE
    except Exception:
        pass

    # 3. Try active_instances.json (user-configured symbol)
    try:
        from config_manager import get_config_dir
        inst_path = get_config_dir() / "active_instances.json"
        if inst_path.exists():
            inst = json.loads(inst_path.read_text(encoding="utf-8"))
            for _k, _v in inst.items():
                if isinstance(_v, dict) and _v.get("symbol"):
                    _DETECTED_SYMBOL_CACHE = str(_v["symbol"]).strip().upper()
                    _DETECTED_SYMBOL_TS = now
                    return _DETECTED_SYMBOL_CACHE
    except Exception:
        pass

    # 4. Fallback
    return "XAUUSD"


def get_price_truth(preferred_symbol: str = "") -> dict[str, Any]:
    if not preferred_symbol:
        preferred_symbol = _detect_active_symbol()
    """Return validated MT5 quote metadata for the requested symbol."""
    import time as _time

    global _last_mt5_error_time
    try:
        _last_mt5_error_time
    except NameError:
        _last_mt5_error_time = 0
    sf = _resolve_mt5_terminal_status_file()
    result = _empty_terminal_quote_result(
        preferred_symbol=preferred_symbol,
        source_path=sf,
    )
    data: dict[str, Any] | None = None

    for _attempt in range(10):
        sf = _resolve_mt5_terminal_status_file()
        result["sourcePath"] = str(sf)
        if sf is None:
            result['reason'] = 'EA_NOT_DEPLOYED'
            result['help'] = 'Run setup.bat as admin, then in MT5 drag GoldAutoTradeBridgeEA onto a chart and enable DLL imports'
            return result
        
        try:
            text_content = sf.read_text(encoding="utf-8", errors="ignore")
            loaded = json.loads(text_content)
            data = loaded if isinstance(loaded, dict) else None
            break
        except Exception:
            _time.sleep(0.05)
    else:
        if _time.time() - _last_mt5_error_time > 60:
            _log(
                f"[ERROR] MT5 terminal_status.json not found in {sf.parent} (Make sure EA is attached to chart)",
                "ERROR",
            )
            _last_mt5_error_time = _time.time()
        result["reason"] = "TERMINAL_STATUS_UNREADABLE"
        result["help"] = "EA Not Connected. Steps: 1) Run setup.bat 2) In MT5, drag GoldAutoTradeBridgeEA onto a chart 3) Enable DLL imports in EA settings 4) Ensure MT5 terminal is running"
        return result

    if not isinstance(data, dict):
        result["reason"] = "TERMINAL_STATUS_INVALID_JSON"
        return result

    normalized_preferred = normalize_symbol(preferred_symbol)
    instruments = (
        data.get("instruments", {})
        if isinstance(data.get("instruments", {}), dict)
        else {}
    )
    root_symbol = normalize_symbol(data.get("symbol", ""), default="")
    result["topLevelSymbol"] = root_symbol

    quote_origin = "invalid"
    quote_symbol = ""
    bid: Optional[float] = None
    ask: Optional[float] = None
    spread: Optional[float] = None
    schema_state = "legacy"
    reason = "MISSING_XAUUSD_QUOTE"

    if instruments:
        schema_state = "instrument_map"
        matched_symbol, payload = pick_best_instrument(instruments, normalized_preferred)
        if isinstance(payload, dict):
            try:
                bid = float(payload.get("bid", 0) or 0)
                ask = float(payload.get("ask", 0) or 0)
                spread = float(payload.get("spread", 0) or 0)
            except Exception:
                bid = ask = spread = None
            if bid and ask:
                quote_origin = f"instrument:{matched_symbol or normalized_preferred}"
                quote_symbol = matched_symbol or normalized_preferred
                reason = ""
            else:
                reason = "XAUUSD_QUOTE_INVALID"
        else:
            reason = "XAUUSD_MISSING_IN_INSTRUMENTS"

    if bid is None or ask is None:
        try:
            root_bid = float(data.get("bid", 0) or 0)
            root_ask = float(data.get("ask", 0) or 0)
            root_spread = float(data.get("spread", 0) or 0)
        except Exception:
            root_bid = root_ask = root_spread = 0.0
        if root_bid > 0 and root_ask > 0:
            if symbol_matches_preferred(root_symbol, normalized_preferred):
                bid = root_bid
                ask = root_ask
                spread = root_spread
                quote_origin = "root"
                quote_symbol = root_symbol or normalized_preferred
                reason = ""
                if schema_state != "instrument_map":
                    schema_state = "root_only"
            else:
                reason = f"ROOT_SYMBOL_MISMATCH:{root_symbol or 'UNKNOWN'}"
        elif not instruments:
            reason = "LEGACY_SCHEMA_NO_QUOTES"

    age_seconds = None
    ea_updated = str(data.get("updated", "") or "").strip()
    if ea_updated:
        try:
            ea_dt = datetime.strptime(ea_updated, "%Y.%m.%d %H:%M:%S")
            sys_dt = datetime.now()
            age_seconds = int((sys_dt - ea_dt).total_seconds())
        except Exception:
            age_seconds = None

    if bid is None or ask is None:
        result.update(
            {
                "reason": reason,
                "schemaState": schema_state,
                "ageSeconds": age_seconds,
            }
        )
        return result

    source = "mt5" if quote_origin == "root" else f"mt5:{quote_origin}"
    if age_seconds is not None and age_seconds > 300:
        source = "stale" if quote_origin == "root" else f"stale:{quote_origin}"

    result.update(
        {
            "ok": True,
            "marketSymbol": quote_symbol or normalized_preferred,
            "bid": bid,
            "ask": ask,
            "spread": spread,
            "quoteOrigin": quote_origin,
            "quoteSymbol": quote_symbol,
            "schemaState": schema_state,
            "reason": "",
            "ageSeconds": age_seconds,
            "source": source,
        }
    )
    return result


def get_price():
    """
    获取实时报价（统一入口 → MT5 Data Hub）。
    返回 (bid, ask, spread, source, age_seconds)
    source: 'mt5' | 'stale' | 'error'
    """
    try:
        from src.mt5_data_hub import hub
        return hub.get_price()
    except Exception:
        pass
    import time as _time
    global _last_mt5_error_time
    try:
        _last_mt5_error_time
    except NameError:
        _last_mt5_error_time = 0

    try:
        truth = get_price_truth()  # auto-detect from EA chart
        if not truth.get("ok", False):
            reason = str(truth.get("reason", "UNKNOWN") or "UNKNOWN")
            source_path = str(truth.get("sourcePath", "") or "")
            if _time.time() - _last_mt5_error_time > 60:
                _log(
                    f"[ERROR] MT5 quote rejected for {_detect_active_symbol()}: reason={reason} path={source_path}",
                    "ERROR",
                )
                _last_mt5_error_time = _time.time()
            return None, None, None, "error", None

        bid = float(truth.get("bid", 0.0) or 0.0)
        ask = float(truth.get("ask", 0.0) or 0.0)
        spread = float(truth.get("spread", 0.0) or 0.0)
        source = str(truth.get("source", "mt5") or "mt5")
        age_seconds = truth.get("ageSeconds")
        if age_seconds is not None and float(age_seconds) > 300:
            _log(f"[WARN] MT5 price stale: {int(age_seconds)}s old (bid={bid})")
        return bid, ask, spread, source, age_seconds or 0
    except Exception as e:
        _log(f"[ERROR] get_price failed: {e}", "ERROR")
        return None, None, None, "error", None

def get_mt5_position_state():
    """读取 MT5 真实持仓状态。

    优先级：
    1. market_snapshot.json（Trading Agent 写入，有真实 MT5 数据）
    2. from_ea/terminal_status.json（EA OnTimer 写入，Wine 下可能陈旧）
    """
    global _STALE_WARN_LAST
    import time as _time

    def _is_snapshot_fresh(snap_data: dict, snap_path, max_age_sec: float = 180.0) -> bool:
        # terminal_status.json format: flat bid/ask, no tick object
        # Check bars[-1].time first — last bar must be within reasonable window.
        # MT5 server time may differ from local system time, so we use max()
        # to handle clock skew, but still enforce an absolute upper bound (30 min
        # for M5 bars) to prevent trading on stale data during MT5 restarts.
        try:
            bars = snap_data.get("bars", [])
            if isinstance(bars, list) and bars:
                last_bar_time = int(bars[-1].get("time", 0) or 0)
                if last_bar_time > 1_000_000_000:
                    age_sec = max(0.0, _time.time() - last_bar_time)
                    # M5 bars: allow up to 30 min age to handle clock skew + gaps.
                    # This catches MT5 restarts while still trusting normal operation.
                    return age_sec <= max(max_age_sec, 1800.0)
        except Exception:
            pass
        # market_snapshot.json format: tick.time_msc — require within max_age
        try:
            tick = snap_data.get("tick", {}) if isinstance(snap_data, dict) else {}
            tick_time_ms = int((tick or {}).get("time_msc", 0) or 0)
            if tick_time_ms > 1_735_000_000_000:
                age_sec = max(0.0, (_time.time() * 1000.0 - tick_time_ms) / 1000.0)
                return age_sec <= max_age_sec
        except Exception:
            pass
        try:
            age_sec = max(0.0, _time.time() - float(snap_path.stat().st_mtime))
            return age_sec <= max_age_sec
        except Exception:
            return False

    # ── 优先从 terminal_status.json 读取（EA OnTimer 60s 写入，最新最可靠）──
    # terminal_status.json 包含实时报价、200 M5 bars、持仓，且 EA 当前版本只写此文件
    snapshot_candidates = []
    try:
        from src.ea_bridge_paths import resolve_bridge_layout
        layout = resolve_bridge_layout(ROOT)
        snapshot_candidates.extend([
            layout["bridge_root"] / "terminal_status.json",
        ])
    except Exception as e:
        _log(f"Failed to resolve bridge layout: {e}", "WARN")

    snapshot_candidates.extend([
        DATA_DIR / "bridge" / "terminal_status.json",
    ])

    # ── 回退到 market_snapshot.json（旧版 EA 输出，可能已过期）──
    try:
        from src.ea_bridge_paths import resolve_bridge_layout
        snapshot_candidates.extend([
            layout["root"] / "market_snapshot.json",
        ])
    except Exception:
        pass

    snapshot_candidates.extend([
        DATA_DIR / "bridge" / "market_snapshot.json",
        ROOT / "bridge" / "market_snapshot.json",
    ])
    for snap_path in snapshot_candidates:
        if snap_path.exists():
            try:
                snap_data = json.loads(snap_path.read_text(encoding='utf-8', errors='ignore'))
                if not _is_snapshot_fresh(snap_data, snap_path):
                    snap_key = str(snap_path)
                    now_ts = _time.time()
                    if now_ts - _STALE_WARN_LAST.get(snap_key, 0) > _STALE_WARN_INTERVAL:
                        _STALE_WARN_LAST[snap_key] = now_ts
                        # DEBUG: log WHY stale detection failed
                        _bars = snap_data.get("bars", [])
                        _last_bar = _bars[-1].get("time", 0) if (_bars and isinstance(_bars, list)) else 0
                        _fmtime = snap_path.stat().st_mtime if snap_path.exists() else 0
                        _daemon_log(
                            f"[MT5_STATE] stale snapshot: bars={len(_bars) if isinstance(_bars,list) else 0} "
                            f"last_bar={_last_bar} age_sec={max(0,_time.time()-_last_bar) if _last_bar>1e9 else 'INVALID'} "
                            f"file_age={_time.time()-_fmtime:.0f}s path={snap_path.name}"
                        )
                    # Auto-cleanup expired market_snapshot.json (not terminal_status.json)
                    if "market_snapshot" in snap_path.name:
                        try:
                            snap_path.unlink()
                            _daemon_log(f"[MT5_STATE] deleted stale snapshot: {snap_path}")
                        except Exception:
                            pass
                    continue
                positions_all = snap_data.get("positions_all", [])
                if not isinstance(positions_all, list):
                    positions_all = []
                legacy_positions = snap_data.get("positions", [])
                if not isinstance(legacy_positions, list):
                    legacy_positions = []
                positions_snapshot = positions_all or legacy_positions
                positions_count = int(snap_data.get("positions_count", snap_data.get("count", 0)) or 0)
                account = snap_data.get("account", {})
                # terminal_status.json has account as a string (login number);
                # market_snapshot.json has account as {equity, balance} dict.
                if isinstance(account, dict):
                    equity = float(account.get("equity", 0) or 0)
                    balance = float(account.get("balance", 0) or 0)
                else:
                    equity = float(snap_data.get("equity", 0) or 0)
                    balance = float(snap_data.get("balance", 0) or 0)
                source_label = "terminal_status" if "terminal_status" in str(snap_path) else "market_snapshot"
                if positions_snapshot:
                    p = positions_snapshot[0]
                    return {
                        "position_open": True,
                        "position_ticket": int(p.get("ticket", 0) or 0),
                        "position_side": str(p.get("side", "BUY") or "BUY").upper(),
                        "position_entry": float(p.get("entry", 0) or 0),
                        "position_price": float(p.get("entry", 0) or 0),
                        "position_sl": float(p.get("sl", 0) or 0),
                        "position_tp": float(p.get("tp", 0) or 0),
                        "position_vol": float(p.get("volume", 0) or 0),
                        "positions_all": positions_snapshot,
                        "balance": balance,
                        "equity": equity,
                        "_source": source_label,
                    }
                if positions_count > 0:
                    _daemon_log(
                        f"[MT5_STATE] {source_label} missing position details despite positions_count={positions_count}; "
                        "falling back to terminal_status"
                    )
                    continue
                else:
                    return {
                        "position_open": False,
                        "positions_all": [],
                        "balance": balance,
                        "equity": equity,
                        "_source": source_label,
                    }
            except Exception:
                pass

    # ── 回退到 terminal_status.json（EA 写入，Wine OnTimer 可能不触发）──
    # 指数退避 + 抖动，防止 EA 写入期间读到半截 JSON 后立即放弃。
    import random as _random
    for attempt in range(8):
        sf = _resolve_mt5_terminal_status_file()
        try:
            text_content = sf.read_text(encoding='utf-8', errors='ignore')
            data = json.loads(text_content)
            balance = float(data.get("balance", 0) or 0)
            equity = float(data.get("equity", 0) or 0)
            positions = data.get("positions", [])
            if not positions:
                return {"position_open": False, "positions_all": [], "balance": balance, "equity": equity}
            p = positions[0]
            return {
                "position_open": True,
                "position_ticket": p.get("ticket", 0),
                "position_side": p.get("side", "BUY"),
                "position_entry": p.get("entry", 0),
                "position_price": p.get("entry", 0),
                "position_sl": p.get("sl", 0),
                "position_tp": p.get("tp", 0),
                "position_vol": p.get("volume", 0),
                "positions_all": positions,
                "balance": balance,
                "equity": equity,
                "_source": "terminal_status",
            }
        except FileNotFoundError:
            if attempt == 7:
                return {"error": "file_missing"}
        except json.JSONDecodeError as e:
            if attempt == 7:
                _log(f"[DEBUG] get_mt5_position_state error: {e}", "DEBUG")
                return {"error": "json_parse_failed"}
        except Exception as e:
            if attempt == 7:
                _log(f"[DEBUG] get_mt5_position_state error: {e}", "DEBUG")
                return {"error": "read_failed"}
            # fall through to sleep+retry

        # Exponential backoff: 0.1s → 0.2s → 0.4s → ... max 1.6s, with ±25% jitter
        delay = min(0.1 * (2 ** attempt), 1.6)
        jitter = delay * 0.25 * (2.0 * _random.random() - 1.0)
        _time.sleep(max(0.02, delay + jitter))
    return {"error": "file_missing"}

def load_state():
    base = {}
    
    # 从 SQLite 加载 KV 状态
    import logging as _log
    init_db()  # 确保表存在（首次运行/升级后）
    try:
        conn = _open_state_db(timeout=10.0)
        c = conn.cursor()
        c.execute("SELECT key, value FROM state_kv")
        for row in c.fetchall():
            try:
                base[row[0]] = json.loads(row[1])
            except:
                base[row[0]] = row[1]
        conn.close()
    except Exception as e:
        _log.warning(f"[gold_trader] load_state() SQLite failed: {e} — falling back to JSON")

    # 兼容老的 JSON 状态文件，如果 SQLite 没数据，从旧文件读并迁移
    if not base and STATE_FILE.exists():
        try:
            base = json.loads(STATE_FILE.read_text(encoding="utf-8"))
            save_state(base) # 写入 SQLite
        except Exception as e:
            _log.warning(f"[gold_trader] load_state() JSON fallback failed: {e} — using defaults")
    
    defaults = {
        "position_open": False, "position_ticket": 0, "position_side": None,
        "position_price": 0.0, "position_sl": 0.0, "position_tp": 0.0,
        "position_vol": 0.0, "position_entry": 0.0,
        "position_signal_id": "",
        "position_decision_id": "",
        "position_regime": "UNKNOWN",
        "position_strategy_id": "",
        "position_trace_id": "",
        "position_confidence": 0.0,
        "position_session_info": {},
        "position_session_metrics": {},
        "position_interpreter": {},
        "position_attribution": {},
        "position_agent_direction": "",
        "position_entry_owner_instance_id": "",
        "position_entry_owner_strategy_id": "",
        "position_current_manager_instance_id": "",
        "position_current_manager_strategy_id": "",
        "position_expected_exit_owner_instance_id": "",
        "position_expected_exit_owner_strategy_id": "",
        "position_exit_owner_instance_id": "",
        "position_exit_owner_strategy_id": "",
        "position_contributors": [],
        "consecutive_losses": 0, "daily_loss_pct": 0.0, "intraday_drawdown_pct": 0.0,
        "account_total_loss_current_pct": 0.0, "account_total_loss_stop_active": False,
        "account_total_loss_stop_threshold_pct": 30.0, "account_total_loss_stop_source": "NONE",
        "total_pnl": 0.0, "win_rate": 0.0,
        "recent_closures": [],
        "confidence_calibration": {},
        "defense_metrics": {},
        "countertrend_budget": {},
        "trailing_telemetry": {},
        "last_closed_side": "",
        "last_closed_ticket": 0,
        "last_closed_ts": 0.0,
        "last_close_action": "",
        "last_signal_id": None, "last_run": None,
        "kill_switch": False, "total_trades": 0,
        "consecutive_timeouts": 0, "last_close_time": 0.0,
        "close_report_dedupe": [],
        "_mt5_sync_err_count": 0,
        "_mt5_sync_forced_close": False,
        "_mt5_sync_forced_close_reason": "",
    }
    for k, v in defaults.items(): base.setdefault(k, v)
    # Reset transient error counters on every load so stale values from a
    # previous run don't prematurely trigger forced-close after restart.
    base["_mt5_sync_err_count"] = 0
    base.pop("_mt5_sync_forced_close", None)
    base.pop("_mt5_sync_forced_close_reason", None)
    # Trim close_report_dedupe on every load — it accumulates when monitor_position_loop
    # runs (process_ea_reports is skipped while position_open=True).
    dedupe = base.get("close_report_dedupe")
    if isinstance(dedupe, list) and len(dedupe) > 200:
        base["close_report_dedupe"] = dedupe[-200:]
    base["win_rate"] = _recent_closure_win_rate(base)
    _sync_consecutive_losses_from_recent(base)  # recalc daily_loss_pct with fixed formula
    if not isinstance(base.get("trailing_telemetry", {}), dict):
        base["trailing_telemetry"] = {}
    if not base.get("trailing_telemetry"):
        cached_trailing = _load_latest_trailing_telemetry()
        if cached_trailing:
            base["trailing_telemetry"] = cached_trailing
    previous_position_open = bool(base.get("position_open", False))
    previous_ticket = int(base.get("position_ticket", 0) or 0)

    persisted_position_snapshot = {
        key: base.get(key)
        for key in base.keys()
        if key.startswith("position_") or key in {"l1_breakeven_done", "partial_done"}
    }
    mt5 = get_mt5_position_state()
    if bool(mt5.get("position_open")) and int(mt5.get("position_ticket", 0) or 0) > 0:
        for k in ("position_open", "position_ticket", "position_side",
                  "position_price", "position_sl", "position_tp", "position_vol", "position_entry", "positions_all",
                  "balance", "equity"):
            if k in mt5:
                base[k] = mt5[k]
        mt5_ticket = int(mt5.get("position_ticket", 0) or 0)
        if previous_position_open and previous_ticket > 0 and previous_ticket != mt5_ticket:
            # A new MT5 ticket means any in-memory runner/breakeven markers belong to
            # the prior position and must not leak into the replacement position.
            # ── De-bounce: 多 MT5 终端的 snapshot 轮询会短暂报告旧/新 ticket，
            # 引入确认窗口：30s 内再次出现相同跳变才算真 rollover；否则只更新 ticket。
            last_rollover_meta = base.get("_last_ticket_rollover") or {}
            if not isinstance(last_rollover_meta, dict):
                last_rollover_meta = {}
            last_seen_pair = (int(last_rollover_meta.get("prev", 0) or 0),
                              int(last_rollover_meta.get("curr", 0) or 0))
            last_seen_ts = float(last_rollover_meta.get("ts", 0.0) or 0.0)
            current_pair = (previous_ticket, mt5_ticket)
            now_ts = time.time()
            if last_seen_pair == current_pair and (now_ts - last_seen_ts) < 30.0:
                # 已确认过的相同 rollover；仅更新 last_seen 时间，避免刷屏日志
                last_rollover_meta["ts"] = now_ts
                base["_last_ticket_rollover"] = last_rollover_meta
            else:
                lifecycle_current = {}
                try:
                    lifecycle_payload = load_position_lifecycle(DATA_DIR)
                    current_payload = (
                        lifecycle_payload.get("current", {})
                        if isinstance(lifecycle_payload.get("current", {}), dict)
                        else {}
                    )
                    lifecycle_status = str(current_payload.get("status", "") or "").strip().lower()
                    lifecycle_ticket = int(current_payload.get("ticket", 0) or 0)
                    if lifecycle_status in {"open", "opening", "closing"} and lifecycle_ticket == previous_ticket:
                        lifecycle_current = current_payload
                except Exception:
                    lifecycle_current = {}

                rollover_attribution = normalize_position_attribution(
                    base.get("position_attribution", {}) if isinstance(base.get("position_attribution", {}), dict) else {},
                    action="OPEN",
                    fallback=lifecycle_current or base,
                )
                base.update(
                    position_state_fields_from_attribution(
                        rollover_attribution,
                        action="OPEN",
                        fallback=lifecycle_current or base,
                    )
                )
                base["l1_breakeven_done"] = False
                base["partial_done"] = False
                base["trend_run_mode"] = False
                base["trailing_telemetry"] = {}
                base["position_time"] = time.time()
                base["_last_ticket_rollover"] = {"prev": previous_ticket, "curr": mt5_ticket, "ts": now_ts}
                try:
                    sync_open_fill(
                        rollover_attribution,
                        data_dir=DATA_DIR,
                        ticket=int(mt5_ticket or 0),
                        side=str(base.get("position_side", mt5.get("position_side", "BUY")) or "BUY").upper(),
                        volume=float(base.get("position_vol", mt5.get("position_vol", 0.0)) or 0.0),
                        entry_price=float(base.get("position_entry", mt5.get("position_entry", 0.0)) or 0.0),
                        sl=float(base.get("position_sl", mt5.get("position_sl", 0.0)) or 0.0),
                        tp=float(base.get("position_tp", mt5.get("position_tp", 0.0)) or 0.0),
                        symbol=str(
                            ((base.get("positions_all", []) or [{}])[0] if isinstance(base.get("positions_all", []), list) and base.get("positions_all", []) else {}).get("symbol", "XAUUSD")
                            or "XAUUSD"
                        ),
                        source="bridge_load_state_ticket_rollover",
                    )
                except Exception as lifecycle_err:
                    _daemon_log(f"sync_open_fill load_state ticket rollover failed: {lifecycle_err}")
                _daemon_log(
                    f"[LOAD_STATE] detected ticket rollover {previous_ticket}->{mt5_ticket}; "
                    "reset runner/breakeven state"
                )
    else:
        # MT5 当前未报告持仓时，显式标为无仓。
        # 只有在实时源不可用时，才保留旧快照给对账器做归因；若实时源已明确返回空仓，
        # 就不能再把 SQLite 里的旧票据/positions_all 回灌成"活跃持仓"。
        base["position_open"] = False
        # Single guard: if MT5 returned a positions_all key (even empty list),
        # it means MT5 is reachable and confirms no open positions — clear all.
        # If positions_all is missing (MT5 unreachable), preserve old snapshot
        # so the reconciler can attribute the ghost later.
        if "positions_all" in mt5:
            base["positions_all"] = mt5["positions_all"]
            if "balance" in mt5:
                base["balance"] = mt5["balance"]
            if "equity" in mt5:
                base["equity"] = mt5["equity"]
            base["position_ticket"] = 0
            base["position_side"] = None
            base["position_price"] = 0.0
            base["position_sl"] = 0.0
            base["position_tp"] = 0.0
            base["position_vol"] = 0.0
            base["position_entry"] = 0.0
            base["l1_breakeven_done"] = False
            base["partial_done"] = False
            base["trend_run_mode"] = False
            base["trailing_telemetry"] = {}
        elif int(persisted_position_snapshot.get("position_ticket", 0) or 0) > 0:
            for key, value in persisted_position_snapshot.items():
                if key == "position_open":
                    continue
                base[key] = value
    _update_account_state(mt5.get("balance"), mt5.get("equity"))

    # 从 effective param 注入最新的 risk_level 给 GUI 显示
    base["risk_level"] = _get_effective_param("risk_level", "balanced")
    total_loss_snapshot = _get_total_loss_stop_snapshot(base)
    base["account_total_loss_current_pct"] = float(total_loss_snapshot.get("current_pct", 0.0) or 0.0)
    base["account_total_loss_stop_active"] = bool(total_loss_snapshot.get("active", False))
    base["account_total_loss_stop_threshold_pct"] = float(total_loss_snapshot.get("threshold_pct", 30.0) or 30.0)
    base["account_total_loss_stop_source"] = str(total_loss_snapshot.get("current_source", "NONE") or "NONE")
    healed = _self_heal_position_state(base)
    if healed:
        try:
            save_state(base)
        except Exception:
            pass
    try:
        # Keep the root runtime blackboard aligned with the reconciled position state.
        _update_shadow_system(
            position_open=bool(base.get("position_open", False)),
            total_trades=base.get("total_trades", 0),
            total_pnl=base.get("total_pnl", 0.0),
            win_rate=base.get("win_rate", 0.0),
        )
        # ── 同步持仓详情到黑板（修复空仓误显问题）──
        try:
            _runtime_blackboard_update("Shadow_System", "position_ticket", int(base.get("position_ticket", 0) or 0))
            _runtime_blackboard_update("Shadow_System", "position_side", str(base.get("position_side") or ""))
            _runtime_blackboard_update("Shadow_System", "position_entry", float(base.get("position_entry", 0.0) or 0.0))
            _runtime_blackboard_update("Shadow_System", "position_price", float(base.get("position_price", 0.0) or 0.0))
            _runtime_blackboard_update("Shadow_System", "position_sl", float(base.get("position_sl", 0.0) or 0.0))
            _runtime_blackboard_update("Shadow_System", "position_tp", float(base.get("position_tp", 0.0) or 0.0))
            _runtime_blackboard_update("Shadow_System", "position_vol", float(base.get("position_vol", 0.0) or 0.0))
        except Exception:
            pass
        _runtime_blackboard_update("Micro_Sniper", "recent_stop_losses", int(base.get("consecutive_losses", 0) or 0))
    except Exception:
        pass
    
    return base


def _heal_open_position_fields_from_positions_all(state: dict, *, emit_log: bool = False) -> bool:
    if not isinstance(state, dict) or not bool(state.get("position_open", False)):
        return False
    positions_all = state.get("positions_all", [])
    if not isinstance(positions_all, list) or not positions_all:
        return False

    active_ticket = int(state.get("position_ticket", 0) or 0)
    snapshot = None
    for item in positions_all:
        if not isinstance(item, dict):
            continue
        if active_ticket > 0 and int(item.get("ticket", 0) or 0) == active_ticket:
            snapshot = item
            break
        if snapshot is None:
            snapshot = item
    if not isinstance(snapshot, dict):
        return False

    changed_fields = []
    snapshot_side = str(snapshot.get("side", "") or "").upper()
    snapshot_entry = float(snapshot.get("entry", 0) or 0.0)
    snapshot_sl = float(snapshot.get("sl", 0) or 0.0)
    snapshot_tp = float(snapshot.get("tp", 0) or 0.0)
    snapshot_vol = float(snapshot.get("volume", snapshot.get("vol", 0)) or 0.0)

    current_side = str(state.get("position_side", "") or "").upper()
    if not current_side and snapshot_side:
        state["position_side"] = snapshot_side
        changed_fields.append("position_side")
    if float(state.get("position_entry", 0) or 0.0) <= 0 and snapshot_entry > 0:
        state["position_entry"] = snapshot_entry
        changed_fields.append("position_entry")
    if float(state.get("position_price", 0) or 0.0) <= 0 and snapshot_entry > 0:
        state["position_price"] = snapshot_entry
        changed_fields.append("position_price")
    if float(state.get("position_sl", 0) or 0.0) <= 0 and snapshot_sl > 0:
        state["position_sl"] = snapshot_sl
        changed_fields.append("position_sl")
    if float(state.get("position_tp", 0) or 0.0) <= 0 and snapshot_tp > 0:
        state["position_tp"] = snapshot_tp
        changed_fields.append("position_tp")
    if float(state.get("position_vol", 0) or 0.0) <= 0 and snapshot_vol > 0:
        state["position_vol"] = snapshot_vol
        changed_fields.append("position_vol")

    if changed_fields and emit_log:
        _daemon_log(
            f"[STATE_HEAL_POSITION] restored {','.join(changed_fields)} from positions_all "
            f"for ticket={active_ticket or int(snapshot.get('ticket', 0) or 0)}"
        )
    return bool(changed_fields)


def _self_heal_position_state(state: dict) -> bool:
    if not isinstance(state, dict):
        return False
    changed = False
    pos_open = bool(state.get("position_open", False))
    ticket = int(state.get("position_ticket", 0) or 0)
    pos_time = float(state.get("position_time", 0) or 0)
    if not pos_open:
        # 当 MT5 侧暂时读不到持仓时，保留本地持仓元数据给 reconcile_positions()
        # 做 ghost/external close 归因。这里如果直接清空，会跳过统一平仓归档。
        return changed
    else:
        if ticket <= 0:
            state.update({
                "position_open": False,
                "position_ticket": 0,
                "position_time": 0,
                "position_side": None,
                "position_entry": 0.0,
                "position_price": 0.0,
                "position_sl": 0.0,
                "position_tp": 0.0,
                "position_vol": 0.0,
            })
            _daemon_log("[STATE_HEAL_POSITION] position_open=True but ticket=0; force reset")
            changed = True
        elif pos_time <= 0:
            state["position_time"] = time.time()
            _daemon_log(f"[STATE_HEAL_POSITION] patched missing position_time for ticket={ticket}")
            changed = True
        if _heal_open_position_fields_from_positions_all(state, emit_log=True):
            changed = True
    return changed


def _sanitize_flat_position_snapshot(state: dict) -> None:
    if not isinstance(state, dict):
        return
    positions_all = state.get("positions_all", [])
    if not isinstance(positions_all, list):
        positions_all = []
    position_open = bool(state.get("position_open", False))
    active_ticket = int(state.get("position_ticket", 0) or 0)
    # `positions_all` is only a derived MT5 snapshot. Once the runtime is flat,
    # keeping old rows here recreates ghost holdings in GUI/open guards.
    if (not position_open) or active_ticket <= 0:
        state["positions_all"] = []
        return
    _heal_open_position_fields_from_positions_all(state, emit_log=False)
    state["positions_all"] = positions_all

def save_state(state):
    try:
        _sanitize_flat_position_snapshot(state)
        # SQLite WAL 模式 + 超时，避免锁死
        conn = _open_state_db(timeout=10.0)
        c = conn.cursor()

        # 将当前的软挂单队列也一并持久化到 state 中，供 GUI 读取
        global _soft_limits
        state["soft_limits"] = _soft_limits

        c.execute("BEGIN TRANSACTION")
        try:
            for k, v in state.items():
                c.execute("INSERT OR REPLACE INTO state_kv (key, value) VALUES (?, ?)", (k, json.dumps(v)))
            conn.commit()
        except Exception:
            conn.rollback()
            raise
        finally:
            conn.close()
        # 同时写 JSON 备援
        _write_json_atomic(STATE_FILE, state)
    except sqlite3.OperationalError:
        # SQLite 只读/锁定: 回退到 JSON 备援
        try:
            _sanitize_flat_position_snapshot(state)
            _write_json_atomic(STATE_FILE, state)
        except Exception:
            pass
    except Exception as e:
        _daemon_log(f"[save_state ERROR] failed to persist state: {e}")


COOLDOWN_SECONDS = 300
TICKET_ZERO_PENDING_TIMEOUT_SEC = 120
BRIDGE_TIMEOUT_PENDING_TIMEOUT_SEC = 120
COUNTERTREND_STRONG_TREND_ADX = 30.0
COUNTERTREND_WEAK_TREND_ADX = 22.0
COUNTERTREND_MAX_TRADES_PER_DAY = 3
COUNTERTREND_MAX_CONSEC_LOSSES = 2
COUNTERTREND_LOSS_BUDGET_ABS = 35.0


def _cooldown_remaining_sec(
    state: dict, cooldown_sec: int = None, strategy_id: str = "", regime: str = ""
) -> int:
    if cooldown_sec is None:
        cooldown_sec = int(config_manager.get_effective_param("post_close_cooldown_sec", 300) or 300)
    return _shared_cooldown_remaining_sec(
        state, cooldown_sec=cooldown_sec, strategy_id=strategy_id, regime=regime
    )


def _is_direction_conflict(agent_dir: str, side: str) -> bool:
    agent = str(agent_dir or "").upper()
    signal_side = str(side or "").upper()
    if agent not in {"SHORT", "LONG"} or signal_side not in {"BUY", "SELL"}:
        return False
    return (agent == "SHORT" and signal_side == "BUY") or (agent == "LONG" and signal_side == "SELL")


def _normalize_agent_override_multiplier(agent_dir: str, side: str, multiplier: float, override_data: dict) -> float:
    side_u = str(side or "").upper()
    agent_u = str(agent_dir or "").upper()
    mult = float(multiplier or 1.0)
    if mult >= 1.0:
        return mult
    if side_u not in {"BUY", "SELL"} or agent_u not in {"LONG", "SHORT"}:
        return mult
    if _is_direction_conflict(agent_u, side_u):
        return mult
    if not isinstance(override_data, dict):
        override_data = {}
    execution_advice = str(override_data.get("execution_advice", "ALLOW_FOLLOW") or "ALLOW_FOLLOW").upper()
    dominant_direction = str(override_data.get("dominant_direction", "") or "").upper()
    continuation_bias = str(override_data.get("continuation_bias", "") or "").upper()
    structural_side = "BUY" if dominant_direction == "BULLISH" or continuation_bias == "BULLISH" else "SELL" if dominant_direction == "BEARISH" or continuation_bias == "BEARISH" else ""
    aligned_floor = 0.90 if execution_advice == "ALLOW_FOLLOW" else 0.85
    if structural_side and structural_side != side_u:
        aligned_floor = min(aligned_floor, 0.80)
    return max(mult, aligned_floor)


def _soft_limit_registered_agent_direction(order: dict) -> str:
    if not isinstance(order, dict):
        return ""
    direct = str(order.get("registered_agent_direction", "") or "").upper()
    if direct:
        return direct
    attribution = order.get("attribution", {}) if isinstance(order.get("attribution", {}), dict) else {}
    return str(attribution.get("agent_direction", "") or "").upper()


def _is_bidirectional_regime(regime: str) -> bool:
    regime_u = normalize_regime_label(regime)
    return is_range_regime(regime_u) or regime_u == "TRANSITION"


def _countertrend_regime_bucket(regime: str, adx: float, trend_strength: str = "") -> str:
    regime_u = normalize_regime_label(regime)
    ts_u = normalize_regime_label(trend_strength)
    adx_v = float(adx or 0.0)
    if regime_u in {"STRONG_TREND", "VOLATILITY_SPIKE", "EXTREME"}:
        return "STRONG_TREND"
    if regime_u in {"WEAK_TREND", "STEALTH_TREND"} and adx_v >= COUNTERTREND_STRONG_TREND_ADX:
        return "STRONG_TREND"
    if regime_u in {"WEAK_TREND", "STEALTH_TREND"} and adx_v >= COUNTERTREND_WEAK_TREND_ADX:
        return "WEAK_TREND"
    if regime_u == "FAKE_TREND":
        return "TRANSITIONAL"
    if _is_bidirectional_regime(regime_u):
        return "RANGE_NEUTRAL"
    return "TRANSITIONAL"


def _is_weak_trend_countertrend_window(regime: str, indicators: dict) -> bool:
    if normalize_regime_label(regime) not in {"WEAK_TREND", "STEALTH_TREND"}:
        return False
    if not isinstance(indicators, dict):
        return False
    adx_v = float(indicators.get("adx", 0.0) or 0.0)
    return COUNTERTREND_WEAK_TREND_ADX <= adx_v < COUNTERTREND_STRONG_TREND_ADX


def _rebuild_countertrend_budget_from_recent(state: dict) -> dict:
    budget = state.get("countertrend_budget", {}) if isinstance(state.get("countertrend_budget", {}), dict) else {}
    rows = state.get("recent_closures", []) if isinstance(state, dict) else []
    if not isinstance(rows, list):
        rows = []
    today = time.strftime("%Y%m%d", time.gmtime())
    trades_used = 0
    loss_used_abs = 0.0
    consec_losses = 0
    for row in rows:
        if not isinstance(row, dict):
            continue
        ts_v = int(row.get("ts", 0) or 0)
        if ts_v <= 0:
            continue
        if time.strftime("%Y%m%d", time.gmtime(ts_v)) != today:
            continue
        if bool(row.get("agent_aligned", True)):
            continue
        trades_used += 1
        try:
            pnl_v = float(row.get("pnl", 0.0) or 0.0)
        except Exception:
            pnl_v = 0.0
        if pnl_v < 0:
            loss_used_abs += abs(pnl_v)
    for row in reversed(rows):
        if not isinstance(row, dict):
            continue
        ts_v = int(row.get("ts", 0) or 0)
        if ts_v <= 0 or time.strftime("%Y%m%d", time.gmtime(ts_v)) != today:
            continue
        if bool(row.get("agent_aligned", True)):
            continue
        try:
            pnl_v = float(row.get("pnl", 0.0) or 0.0)
        except Exception:
            pnl_v = 0.0
        if pnl_v < 0:
            consec_losses += 1
        else:
            break
    budget.update(
        {
            "date": today,
            "trades_used": int(trades_used),
            "loss_used_abs": round(float(loss_used_abs), 2),
            "consec_losses": int(consec_losses),
            "updated_at": int(time.time()),
        }
    )
    state["countertrend_budget"] = budget
    return budget


def _evaluate_countertrend_quality(signal: dict, bid: float, ask: float) -> dict:
    ind = signal.get("indicators", {}) if isinstance(signal.get("indicators", {}), dict) else {}
    interpreter = signal.get("interpreter", {}) if isinstance(signal.get("interpreter", {}), dict) else {}
    side = str(signal.get("side", "") or "").upper()
    event_tag = str(ind.get("event", "") or "").upper()
    conf = float(signal.get("confidence", 0.0) or 0.0)
    adx_v = float(ind.get("adx", 0.0) or 0.0)
    atr_v = max(0.01, float(ind.get("atr14", 0.0) or 0.0))
    entry = float(signal.get("entry", bid if side == "SELL" else ask) or (bid if side == "SELL" else ask))
    sl = float(signal.get("sl", 0.0) or 0.0)
    tp = float(signal.get("tp", 0.0) or 0.0)
    risk_dist = abs(entry - sl)
    reward_dist = abs(tp - entry)
    rr = (reward_dist / risk_dist) if risk_dist > 1e-6 else 0.0
    spread_to_atr = max(0.0, float(ask - bid)) / atr_v
    struct_score = 0.0
    if (side == "BUY" and event_tag == "BOS_UP") or (side == "SELL" and event_tag == "BOS_DOWN"):
        struct_score += 0.45
    confirmation = str(interpreter.get("confirmation_state", "NONE") or "NONE").upper()
    pullback = str(interpreter.get("pullback_quality", "UNKNOWN") or "UNKNOWN").upper()
    if confirmation == "CONFIRMED":
        struct_score += 0.25
    if pullback in {"HEALTHY", "STRONG"}:
        struct_score += 0.20
    if adx_v >= COUNTERTREND_WEAK_TREND_ADX:
        struct_score += 0.10
    # EMA 三线排列强结构加分：当信号方向与 EMA9>21>50 或 9<21<50 同向，
    # 视为强趋势结构证据（即使无 BOS 事件）
    ema9 = float(ind.get("ema9", 0.0) or 0.0)
    ema21 = float(ind.get("ema21", 0.0) or 0.0)
    ema50 = float(ind.get("ema50", 0.0) or 0.0)
    if ema9 and ema21 and ema50:
        if side == "BUY" and ema9 > ema21 > ema50:
            struct_score += 0.30
        elif side == "SELL" and ema9 < ema21 < ema50:
            struct_score += 0.30
    struct_score = max(0.0, min(1.0, struct_score))
    quality_score = max(
        0.0,
        min(
            1.0,
            0.35 * max(0.0, min(1.0, (conf - 0.58) / 0.26))
            + 0.30 * max(0.0, min(1.0, (rr - 1.05) / 0.85))
            + 0.25 * struct_score
            + 0.10 * max(0.0, min(1.0, (0.30 - spread_to_atr) / 0.30)),
        ),
    )
    return {
        "rr": round(rr, 3),
        "spread_to_atr": round(spread_to_atr, 3),
        "structure_score": round(struct_score, 3),
        "quality_score": round(quality_score, 3),
        "confidence": round(conf, 3),
    }


def _apply_countertrend_budget_gate(state: dict, signal: dict, agent_dir: str, bid: float, ask: float, force_macro_poll: bool = True) -> dict:
    # [DEPRECATED] Pipeline manages counter-trend via signal_arbitrator and macro direction weights.
    # This legacy counter-trend budget system is retired.
    return {"allow": True, "countertrend": False, "reason": "RETIRED_PIPELINE_MANAGES"}
    ind = signal.get("indicators", {}) if isinstance(signal.get("indicators", {}), dict) else {}
    regime_u = str(signal.get("regime", "UNKNOWN") or "UNKNOWN").upper()
    adx_v = float(ind.get("adx", 0.0) or 0.0)
    trend_strength = str(ind.get("trend_strength", "") or "")
    bucket = _countertrend_regime_bucket(regime_u, adx_v, trend_strength)
    quality = _evaluate_countertrend_quality(signal, bid, ask)
    budget = _rebuild_countertrend_budget_from_recent(state)
    session_name = str((signal.get("session_metrics", {}) or {}).get("session_name", "") or "")
    max_trades = int(_get_effective_param("countertrend_daily_max_trades", COUNTERTREND_MAX_TRADES_PER_DAY) or COUNTERTREND_MAX_TRADES_PER_DAY)
    max_consec = int(_get_effective_param("countertrend_max_consecutive_losses", COUNTERTREND_MAX_CONSEC_LOSSES) or COUNTERTREND_MAX_CONSEC_LOSSES)
    max_loss_abs = float(_get_effective_param("countertrend_daily_loss_budget_abs", COUNTERTREND_LOSS_BUDGET_ABS) or COUNTERTREND_LOSS_BUDGET_ABS)
    if session_name == "ASIA":
        max_trades = max(1, max_trades - 1)
    elif session_name == "LONDON_NY_OVERLAP":
        max_trades = min(max_trades + 1, max_trades + 1)
    if bucket == "STRONG_TREND":
        allow = False
        reason = "COUNTERTREND_STRONG_TREND_BLOCK"
    else:
        rr_floor = 1.25 if bucket in {"RANGE_NEUTRAL", "TRANSITIONAL"} else 1.40
        scoring = config_manager.get_scoring_floors()
        if bucket in {"RANGE_NEUTRAL", "TRANSITIONAL"}:
            conf_floor = scoring["ranging"]["conf_floor"]
            struct_floor = scoring["ranging"]["struct_floor"]
        else:
            conf_floor = scoring["trending"]["conf_floor"]
            struct_floor = scoring["trending"]["struct_floor"]
        spread_floor = 0.26 if bucket in {"RANGE_NEUTRAL", "TRANSITIONAL"} else 0.20
        # EMA 三线排列减免：当信号方向与 EMA9>21>50 或 9<21<50 同向时，
        # EMA 排列本身就是结构证据，降低 conf/struct 阈值
        _ind_ema = signal.get("indicators", {}) if isinstance(signal.get("indicators", {}), dict) else {}
        _e9 = float(_ind_ema.get("ema9", 0.0) or 0.0)
        _e21 = float(_ind_ema.get("ema21", 0.0) or 0.0)
        _e50 = float(_ind_ema.get("ema50", 0.0) or 0.0)
        if _e9 and _e21 and _e50 and ((side == "BUY" and _e9 > _e21 > _e50) or (side == "SELL" and _e9 < _e21 < _e50)):
            conf_floor = max(0.56, conf_floor - 0.08)
            struct_floor = max(0.40, struct_floor - 0.15)
        allow = (
            quality["rr"] >= rr_floor
            and quality["confidence"] >= conf_floor
            and quality["structure_score"] >= struct_floor
            and quality["spread_to_atr"] <= spread_floor
            and int(budget.get("trades_used", 0) or 0) < max_trades
            and int(budget.get("consec_losses", 0) or 0) < max_consec
            and float(budget.get("loss_used_abs", 0.0) or 0.0) < max_loss_abs
        )
        reason = "COUNTERTREND_ALLOWED" if allow else "COUNTERTREND_BUDGET_OR_QUALITY_BLOCK"
    high_quality_blocked = (not allow) and quality["quality_score"] >= 0.74
    blocked_streak = int(budget.get("blocked_hq_streak", 0) or 0)
    if high_quality_blocked:
        blocked_streak += 1
    else:
        blocked_streak = 0
    budget["blocked_hq_streak"] = blocked_streak
    budget["last_eval"] = {
        "bucket": bucket,
        "reason": reason,
        "quality": quality,
        "allow": bool(allow),
        "at": int(time.time()),
    }
    state["countertrend_budget"] = budget
    if high_quality_blocked and force_macro_poll and blocked_streak >= 2:
        last_poll = float(budget.get("last_hq_block_macro_poll_ts", 0.0) or 0.0)
        quick_poll_cooldown = float(_get_effective_param("countertrend_macro_repoll_cooldown_sec", 300) or 300)
        if (time.time() - last_poll) >= max(120.0, quick_poll_cooldown):
            _ct_pos = None
            if state.get("position_open") and state.get("position_side"):
                _ct_pos = {
                    "side": state["position_side"],
                    "entry_price": state.get("position_entry", 0),
                    "sl": state.get("position_sl", 0),
                    "tp": state.get("position_tp", 0),
                    "age_sec": time.time() - float(state.get("position_time", time.time())),
                }
            _ct_regime = str((signal.get("indicators", {}) or {}).get("regime", "UNKNOWN") or "UNKNOWN")
            if _trigger_macro_direction_poll("COUNTERTREND_HQ_BLOCKED_STREAK", force=True, current_position=_ct_pos, current_regime=_ct_regime):
                budget["last_hq_block_macro_poll_ts"] = time.time()
                budget["blocked_hq_streak"] = 0
                state["countertrend_budget"] = budget
    return {
        "allow": bool(allow),
        "countertrend": True,
        "bucket": bucket,
        "reason": reason,
        "quality": quality,
        "budget": {
            "trades_used": int(budget.get("trades_used", 0) or 0),
            "max_trades": int(max_trades),
            "loss_used_abs": float(budget.get("loss_used_abs", 0.0) or 0.0),
            "max_loss_abs": float(max_loss_abs),
            "consec_losses": int(budget.get("consec_losses", 0) or 0),
            "max_consec_losses": int(max_consec),
        },
    }


def _is_overlap_window(dt: datetime.datetime = None) -> bool:
    """判断当前是否在欧美重叠时段 (13:00-17:00 UTC)"""
    now_dt = dt or datetime.datetime.now(datetime.timezone.utc)
    return 13 <= now_dt.hour < 17


def _current_hour_key(dt: datetime.datetime = None) -> str:
    now_dt = dt or datetime.datetime.now(datetime.timezone.utc)
    return now_dt.strftime("%Y%m%d%H")


def _get_overlap_attempts(state: dict, dt: datetime.datetime = None) -> int:
    budget = state.get("opportunity_budget", {}) if isinstance(state.get("opportunity_budget", {}), dict) else {}
    slot = budget.get(_current_hour_key(dt), {})
    if not isinstance(slot, dict):
        return 0
    return int(slot.get("effective_attempts", 0) or 0)


def _add_overlap_attempt(state: dict, reason: str = ""):
    now_dt = datetime.datetime.now(datetime.timezone.utc)
    budget = state.get("opportunity_budget", {}) if isinstance(state.get("opportunity_budget", {}), dict) else {}
    key = _current_hour_key(now_dt)
    slot = budget.get(key, {}) if isinstance(budget.get(key, {}), dict) else {}
    attempts = int(slot.get("effective_attempts", 0) or 0) + 1
    budget[key] = {
        "effective_attempts": attempts,
        "last_reason": reason,
        "updated_at": int(time.time()),
    }
    # 仅保留最近 8 小时桶，避免 state 膨胀。
    keys = sorted(budget.keys())
    if len(keys) > 8:
        for old_key in keys[:-8]:
            budget.pop(old_key, None)
    state["opportunity_budget"] = budget


def _set_final_reject_reason(state: dict, reason: str, stage: str = "unknown"):
    state["last_final_reject"] = {
        "reason": str(reason or "UNKNOWN"),
        "stage": str(stage or "unknown"),
        "ts": int(time.time()),
    }


def _append_decision_link(
    *,
    signal_id: str,
    decision_id: str,
    action: str = "",
    stable_id: str = "",
    ticket: int = 0,
    source: str = "",
    request_origin: str = "",
    legacy_fallback: bool = False,
    intent_id: str = "",
) -> None:
    try:
        if not signal_id or not decision_id:
            return
        DECISION_LINK_FILE.parent.mkdir(parents=True, exist_ok=True)
        with open(DECISION_LINK_FILE, "a", encoding="utf-8") as f:
            f.write(
                json.dumps(
                    {
                        "ts": time.strftime("%Y-%m-%dT%H:%M:%SZ"),
                        "signal_id": str(signal_id),
                        "decision_id": str(decision_id),
                        "action": str(action or ""),
                        "stable_id": str(stable_id or ""),
                        "ticket": int(ticket or 0),
                        "source": str(source or ""),
                        "request_origin": str(request_origin or ""),
                        "legacy_fallback": bool(legacy_fallback),
                        "intent_id": str(intent_id or ""),
                    },
                    ensure_ascii=False,
                )
                + "\n"
            )
    except Exception as e:
        _daemon_log(f"append decision link failed: {e}")


def _resolve_decision_id(signal_id=None, ticket=0):
    try:
        if not DECISION_LINK_FILE.exists():
            return ""
        lines = DECISION_LINK_FILE.read_text(encoding="utf-8").splitlines()
        for line in reversed(lines[-500:]):
            if not line.strip():
                continue
            row = json.loads(line)
            if signal_id and str(row.get("signal_id", "")) == str(signal_id):
                return str(row.get("decision_id", ""))
            if ticket and int(row.get("ticket", 0) or 0) == int(ticket):
                return str(row.get("decision_id", ""))
    except Exception as e:
        _daemon_log(f"resolve decision link failed: {e}")
    return ""


def _confirm_open_from_mt5(
    state: dict,
    *,
    decision_id: str = "",
    signal_id: str = "",
    report: Optional[dict] = None,
    pending_reason: str = "MT5_POSITION_CONFIRMED",
):
    """When EA reports a ghost fill (ticket=0), recover the real open from MT5 state."""
    try:
        mt5_state = get_mt5_position_state()
    except Exception as e:
        _daemon_log(f"[PENDING_CONFIRM] mt5 state read failed: {e}")
        return state, False

    if mt5_state.get("error"):
        return state, False

    mt5_ticket = int(mt5_state.get("position_ticket", 0) or 0)
    mt5_entry = float(mt5_state.get("position_entry", 0) or 0)
    if not mt5_state.get("position_open") or mt5_ticket <= 0 or mt5_entry <= 0:
        return state, False

    report_payload = report if isinstance(report, dict) else {}
    report_attribution = report_payload.get("attribution", {}) if isinstance(report_payload.get("attribution", {}), dict) else {}
    pending_attribution = state.get("position_attribution", {}) if isinstance(state.get("position_attribution", {}), dict) else {}
    execution_attribution = normalize_position_attribution(
        report_attribution or pending_attribution,
        action="OPEN",
        fallback=pending_attribution,
    )
    pipeline_owned = _report_owned_by_pipeline(report_payload) or str(execution_attribution.get("writer_owner", "") or "").lower() == "pipeline"

    resolved_decision_id = str(
        decision_id
        or report_payload.get("decision_id")
        or state.get("position_decision_id", "")
        or _resolve_decision_id(ticket=mt5_ticket)
        or _resolve_decision_id(signal_id=signal_id)
        or ""
    ).strip()
    resolved_signal_id = str(
        signal_id
        or report_payload.get("signal_id")
        or state.get("position_signal_id", "")
        or resolved_decision_id
        or ""
    ).strip()

    if report_payload:
        report_fill_price = float(report_payload.get("fill_price", 0) or 0)
        report_volume = float(report_payload.get("volume", 0.0) or 0.0)
        report_side = str(report_payload.get("side", "") or "").upper()
    else:
        report_fill_price = 0.0
        report_volume = 0.0
        report_side = ""

    was_open = bool(state.get("position_open"))
    prev_ticket = int(state.get("position_ticket", 0) or 0)
    next_open_state = {
        "position_open": True,
        "position_side": mt5_state.get("position_side") or report_side or "BUY",
        "position_entry": mt5_entry or report_fill_price,
        "position_price": float(mt5_state.get("position_price", mt5_entry) or mt5_entry or report_fill_price),
        "position_sl": float(mt5_state.get("position_sl", 0) or report_payload.get("sl_price", 0) or 0),
        "position_tp": float(mt5_state.get("position_tp", 0) or report_payload.get("tp_price", 0) or 0),
        "position_vol": float(mt5_state.get("position_vol", 0) or report_volume or 0.01),
        "position_ticket": mt5_ticket,
        "position_signal_id": resolved_signal_id,
        "position_decision_id": resolved_decision_id or resolved_signal_id,
        "position_regime": execution_attribution.get("regime", state.get("position_regime", "UNKNOWN")),
        "position_confidence": float(
            report_payload.get("confidence", state.get("position_confidence", 0.0))
            or state.get("position_confidence", 0.0)
            or 0.0
        ),
        "position_time": time.time(),
    }
    next_open_state.update(position_state_fields_from_attribution(execution_attribution, action="OPEN"))

    _update_account_state(mt5_state.get("balance"), mt5_state.get("equity"))
    state.update(next_open_state)
    state["l1_breakeven_done"] = False
    state["partial_done"] = False

    pending_map = state.get("ticket_zero_pending", {}) if isinstance(state.get("ticket_zero_pending", {}), dict) else {}
    if resolved_decision_id:
        pending_map.pop(resolved_decision_id, None)
    state["ticket_zero_pending"] = pending_map

    is_new_open = (not was_open) or prev_ticket != mt5_ticket
    if is_new_open:
        state["total_trades"] = int(state.get("total_trades", 0) or 0) + 1
        _log(
            f"[OK] POSITION OPENED: {state['position_side']} @{state['position_entry']} "
            f"SL={state['position_sl']} TP={state['position_tp']} [confirmed by MT5]",
            "WARN",
        )
        if not pipeline_owned:
            _log_open(state)
            _update_shadow_system(position_open=True, total_trades=state.get("total_trades", 0), total_pnl=state.get("total_pnl", 0.0))
        try:
            queued_post_fill_repair = _queue_post_fill_sl_tp_repair(
                ticket=int(state.get("position_ticket", 0) or 0),
                side=str(state.get("position_side", "") or ""),
                fill_price=float(state.get("position_entry", 0) or 0),
                sl=float(state.get("position_sl", 0) or 0),
                tp=float(state.get("position_tp", 0) or 0),
                volume=float(state.get("position_vol", 0.0) or 0.0),
                decision_id=str(resolved_decision_id or state.get("position_decision_id", "") or ""),
                signal_id=str(resolved_signal_id or state.get("position_signal_id", "") or ""),
                attribution=execution_attribution,
            )
            if queued_post_fill_repair:
                state["position_post_fill_rr_repair_pending"] = True
        except Exception as post_fill_err:
            _daemon_log(f"[POST_FILL_RR_REPAIR] failed: {post_fill_err}")
        try:
            sys.path.insert(0, str(ROOT))
            from skill_trading_memory import save_position_snapshot
            save_position_snapshot(positions=[{
                "ticket": state["position_ticket"],
                "side": state["position_side"],
                "volume": state["position_vol"],
                "entry": state["position_entry"],
                "sl": state["position_sl"],
                "tp": state["position_tp"],
            }])
        except Exception as e:
            _daemon_log(f"save_position_snapshot failed: {e}")

    if resolved_decision_id and not pipeline_owned:
        _close_decision_state(
            resolved_decision_id,
            "CONFIRMED",
            reason=pending_reason,
            ticket=state.get("position_ticket", 0),
            signal_id=resolved_signal_id,
            side=state.get("position_side"),
            fill_price=state.get("position_entry", 0),
            volume=state.get("position_vol", 0),
        )
        _append_decision_link(
            signal_id=resolved_signal_id,
            decision_id=resolved_decision_id,
            action="OPEN",
            ticket=state.get("position_ticket", 0),
            source="mt5_pending_confirmed",
        )
        try:
            execution_report = dict(report_payload) if report_payload else {}
            execution_report.setdefault("signal_id", resolved_signal_id)
            execution_report.setdefault("decision_id", resolved_decision_id)
            execution_report.setdefault("status", "FILLED")
            execution_report.setdefault("action", "OPEN")
            execution_report.setdefault("ticket", state.get("position_ticket", 0))
            execution_report.setdefault("fill_price", state.get("position_entry", 0))
            execution_report.setdefault("sl_price", state.get("position_sl", 0))
            execution_report.setdefault("tp_price", state.get("position_tp", 0))
            execution_report.setdefault("volume", state.get("position_vol", 0))
            execution_report.setdefault("side", state.get("position_side", "BUY"))
            execution_report.setdefault("attribution", execution_attribution)
            _record_learning_execution(resolved_decision_id, execution_report, state)
        except Exception as learning_err:
            _daemon_log(f"_record_learning_execution pending confirm failed: {learning_err}")
    if not pipeline_owned:
        try:
            sync_open_fill(
                execution_attribution,
                data_dir=DATA_DIR,
                ticket=int(state.get("position_ticket", 0) or 0),
                side=str(state.get("position_side", "BUY") or "BUY"),
                volume=float(state.get("position_vol", 0.0) or 0.0),
                entry_price=float(state.get("position_entry", 0.0) or 0.0),
                sl=float(state.get("position_sl", 0.0) or 0.0),
                tp=float(state.get("position_tp", 0.0) or 0.0),
                symbol="XAUUSD",
                source="bridge_pending_confirm",
            )
        except Exception as lifecycle_err:
            _daemon_log(f"sync_open_fill pending confirm failed: {lifecycle_err}")

    return state, True


def _expire_bridge_timeout_pending(state: dict, decision_id: str, *, timeout_sec: int) -> None:
    _close_decision_state(
        decision_id,
        "FAILED",
        reason="EA_BRIDGE_TIMEOUT_FINAL",
        timeout_sec=timeout_sec,
    )


def _reconcile_bridge_timeout_pending(state: dict) -> dict:
    pending_map = state.get("bridge_timeout_pending", {}) if isinstance(state.get("bridge_timeout_pending", {}), dict) else {}
    if not pending_map:
        state["bridge_timeout_pending"] = {}
        return state

    now_ts = time.time()
    next_pending: dict[str, dict[str, Any]] = {}
    for decision_id, payload in pending_map.items():
        payload_dict = payload if isinstance(payload, dict) else {}
        signal_id = str(payload_dict.get("signal_id", "") or "").strip()
        created_ts = float(payload_dict.get("created_ts", 0.0) or 0.0)
        state, rescued = _confirm_open_from_mt5(
            state,
            decision_id=str(decision_id),
            signal_id=signal_id,
            pending_reason="EA_BRIDGE_TIMEOUT_MT5_CONFIRMED",
        )
        if rescued:
            continue
        age = max(0.0, now_ts - created_ts) if created_ts > 0 else (BRIDGE_TIMEOUT_PENDING_TIMEOUT_SEC + 1)
        if age >= BRIDGE_TIMEOUT_PENDING_TIMEOUT_SEC:
            _expire_bridge_timeout_pending(state, str(decision_id), timeout_sec=BRIDGE_TIMEOUT_PENDING_TIMEOUT_SEC)
            continue
        next_pending[str(decision_id)] = payload_dict

    state["bridge_timeout_pending"] = next_pending
    return state

# ============ Agent 大方向决策 ============
# Agent 的核心职责：定当天大方向（LONG / SHORT / NEUTRAL）
# 规则层只在该方向内执行，不允许逆势开仓

_DIRECTION_V2_CACHE = {
    "ts": 0.0,
    "direction": "NEUTRAL",
    "data": {},
    "last_flip_ts": 0.0,
    "pending_dir": "",
    "pending_count": 0,
}
_DIRECTION_V2_CACHE_TTL = 20.0  # 缓存有效期（秒），避免短时间内重复计算


def _ema_seq(values, period):
    if not values or len(values) < period:
        return []
    k = 2.0 / (period + 1.0)
    out = [sum(values[:period]) / period]
    for v in values[period:]:
        out.append(v * k + out[-1] * (1.0 - k))
    return out


def _sigmoid(x):
    x = max(-20.0, min(20.0, x))
    return 1.0 / (1.0 + math.exp(-x))


def _direction_from_bars(bars):
    closes = [float(b.get("close", 0.0) or 0.0) for b in bars if b.get("close") is not None]
    if len(closes) < 60:
        return 0.0, {}
    ema9 = _ema_seq(closes, 9)
    ema21 = _ema_seq(closes, 21)
    ema50 = _ema_seq(closes, 50)
    if not ema9 or not ema21 or not ema50:
        return 0.0, {}

    e9 = ema9[-1]
    e21 = ema21[-1]
    e50 = ema50[-1]
    trend = 1.0 if (e9 > e21 > e50) else -1.0 if (e9 < e21 < e50) else 0.0
    e21_prev = ema21[-4] if len(ema21) >= 4 else ema21[0]
    slope = (e21 - e21_prev) / max(abs(e21_prev) * 0.0002, 0.05)
    spread = (e9 - e21) / max(abs(e21) * 0.00015, 0.05)
    slope_score = math.tanh(slope)
    spread_score = math.tanh(spread)
    score = max(-1.0, min(1.0, 0.6 * trend + 0.25 * slope_score + 0.15 * spread_score))
    return score, {
        "ema9": round(e9, 3),
        "ema21": round(e21, 3),
        "ema50": round(e50, 3),
        "trend": trend,
        "slope_score": round(slope_score, 4),
        "spread_score": round(spread_score, 4),
        "score": round(score, 4),
    }


def _compute_direction_v2():
    try:
        from bridge.strategy import _read_bars
    except Exception:
        return "NEUTRAL", {"engine": "direction_v2", "reason": "read_bars_unavailable", "probability": 0.5}

    bars_m5 = _read_bars(220, timeframe="M5") or []
    bars_m15 = _read_bars(220, timeframe="M15") or []
    bars_h1 = _read_bars(220, timeframe="H1") or []
    if len(bars_m5) < 80 or len(bars_m15) < 80 or len(bars_h1) < 80:
        return "NEUTRAL", {"engine": "direction_v2", "reason": "insufficient_bars", "probability": 0.5}

    s5, d5 = _direction_from_bars(bars_m5)
    s15, d15 = _direction_from_bars(bars_m15)
    s1h, d1h = _direction_from_bars(bars_h1)
    raw = 0.45 * s5 + 0.35 * s15 + 0.20 * s1h

    # 结构突破增益（5m）
    highs = [float(b.get("high", 0.0) or 0.0) for b in bars_m5]
    lows = [float(b.get("low", 0.0) or 0.0) for b in bars_m5]
    close_last = float(bars_m5[-1].get("close", 0.0) or 0.0)
    if len(highs) > 25 and close_last > max(highs[-21:-1]):
        raw += 0.18
        structure_event = "BOS_UP"
    elif len(lows) > 25 and close_last < min(lows[-21:-1]):
        raw -= 0.18
        structure_event = "BOS_DOWN"
    else:
        structure_event = "NONE"

    # ADX 作为趋势可信度缩放，直接从 bars 计算，避免遗留 EMA 预热链污染运行态
    adx_v = 0.0
    try:
        from src.regime_router import _compute_adx_di
        adx_v = float(_compute_adx_di(bars_m5).get("adx", 0.0) or 0.0)
    except Exception:
        adx_v = 0.0
    adx_scale = 0.72 + 0.38 * max(0.0, min(1.0, (adx_v - 18.0) / 20.0))
    raw *= adx_scale

    p_long = _sigmoid(raw * 2.6)
    # 方向离散化：尊重 allow_short 配置，允许做空
    if raw >= 0.18:
        direction = "LONG"
    elif raw <= -0.18:
        direction = "SHORT"  # 允许做空 — allow_short 在上层 gate 控制
    else:
        direction = "NEUTRAL"
    probability = max(p_long, 1.0 - p_long)
    probability = max(0.5, min(0.98, probability))

    phase = "EARLY"
    if adx_v >= 35 and abs(raw) >= 0.45:
        phase = "MIDDLE"
    if abs(raw) < 0.22 or adx_v < 20:
        phase = "LATE"

    data = {
        "engine": "direction_v2",
        "direction": direction,
        "probability": round(probability, 4),
        "phase": phase,
        "raw_score": round(raw, 4),
        "adx": round(adx_v, 2),
        "structure_event": structure_event,
        "components": {"m5": d5, "m15": d15, "h1": d1h},
        "updated": time.strftime("%Y-%m-%dT%H:%M:%SZ"),
        "updated_ts": time.time(),
    }
    return direction, data


def _load_macro_direction_record():
    """宏观方向唯一入口：只读取 M2（大模型）落盘结果，不允许技术核实时改写。"""
    contract_payload = load_macro_contract_payload()
    if contract_payload:
        return str(contract_payload.get("direction", "NEUTRAL") or "NEUTRAL").upper(), contract_payload
    return _registry_load_macro_direction_record()

def get_agent_direction():
    """读取 Agent 当天的方向决策。返回: LONG / SHORT / NEUTRAL

    优先级: Macro_Commander > contract > registry > NEUTRAL
    Macro_Commander 直接连接 LLM，方向判断最准确。
    """
    # ── Primary: Macro_Commander blackboard (read state file directly) ──
    try:
        import json as _json_bb, os as _os_bb
        # Try all paths where blackboard_state.json may live:
        #  - data/bridge/state/ (global_blackboard from swarm — macro refresh writes here)
        #  - project root (gold_trader + pipeline)
        #  - logs/ (some deployments)
        for _bb_dir in [
            APP_DIR / "data" / "bridge" / "state",
            APP_DIR,
            config_manager.get_logs_dir(),
        ]:
            _bb_path = _bb_dir / "blackboard_state.json"
            if not _bb_path.exists():
                continue
            _bb = _json_bb.loads(_bb_path.read_text(encoding="utf-8"))
            mc = _bb.get("Macro_Commander", {}) if isinstance(_bb, dict) else {}
            mc_dir = str(mc.get("bias", "") or "").upper()
            mc_conf = float(mc.get("confidence", 0.0) or 0.0)
            if not mc_dir:
                continue
            mc_ts = float(mc.get("ts", mc.get("updated_ts", 0)) or 0)
            return mc_dir, {"direction": mc_dir, "engine": "macro_commander",
                            "confidence": mc_conf, "reason": str(mc.get("trigger_reason", "")),
                            "source": "blackboard",
                            "updated_ts": mc_ts,
                            "macro_age_sec": max(0.0, time.time() - mc_ts) if mc_ts > 0 else 99999,
                            "degraded": bool(mc.get("degraded", False))}
    except Exception:
        pass

    # ── Fallback: contract ──
    contract_payload = load_macro_contract_payload()
    if contract_payload:
        direction = str(contract_payload.get("direction", "NEUTRAL") or "NEUTRAL").upper()
        if direction in {"LONG", "SHORT"}:
            contract_ts = float(contract_payload.get("updated_ts", contract_payload.get("updated_at", 0)) or 0)
            return direction, {
                **contract_payload,
                "engine": "macro_contract",
                "updated_ts": contract_ts,
                "macro_age_sec": max(0.0, time.time() - contract_ts) if contract_ts > 0 else 99999,
                "degraded": bool(contract_payload.get("degraded", False)),
                "source": "contract",
            }

    # ── Fallback: registry ──
    reg_dir, reg_data = _registry_get_agent_direction()
    if reg_dir in {"LONG", "SHORT"}:
        return reg_dir, reg_data

    return "NEUTRAL", reg_data if reg_data else {}


def get_technical_trend_bias(force_refresh: bool = False):
    """
    只读技术趋势偏向，用于 GUI 展示。
    不参与主循环宏观方向裁决，也不改写宏观落盘结果。
    """
    now_ts = time.time()
    cache_ttl = _DIRECTION_V2_CACHE_TTL
    if (
        not force_refresh
        and _DIRECTION_V2_CACHE.get("data")
        and (now_ts - float(_DIRECTION_V2_CACHE.get("ts", 0.0) or 0.0)) < cache_ttl
    ):
        return _DIRECTION_V2_CACHE.get("direction", "NEUTRAL"), dict(_DIRECTION_V2_CACHE.get("data", {}))

    direction, data = _compute_direction_v2()
    _DIRECTION_V2_CACHE["ts"] = now_ts
    _DIRECTION_V2_CACHE["direction"] = direction
    _DIRECTION_V2_CACHE["data"] = data if isinstance(data, dict) else {}
    return direction, dict(_DIRECTION_V2_CACHE.get("data", {}))


def _spawn_macro_direction_process(
    force_refresh: bool = False,
    reason: str = "",
    current_position: dict = None,
    current_regime: str = "UNKNOWN",
) -> bool:
    """[DEPRECATED] Pipeline's _try_macro_direction_refresh() manages macro direction.

    This legacy function is retired. Pipeline calls Commander directly via
    _try_macro_direction_refresh() in the main polling loop.
    """
    _daemon_log("[DEPRECATED] _spawn_macro_direction_process called — "
                "pipeline._try_macro_direction_refresh() is the single macro direction path.", "WARN")
    return False


def _trigger_macro_direction_poll(reason: str, force: bool = False, current_position: dict = None, current_regime: str = "UNKNOWN") -> bool:
    """[DEPRECATED] Pipeline's _try_macro_direction_refresh() manages macro direction independently."""
    _daemon_log("[DEPRECATED] _trigger_macro_direction_poll called — "
                "pipeline._try_macro_direction_refresh() is the single macro direction path.", "WARN")
    return False


def _resolve_direction_conflict(agent_dir: str, direction_data: dict, signal: dict, indicators: dict, state: dict):
    """[DEPRECATED] Pipeline resolves direction conflicts via signal_arbitrator + macro authority."""
    return {"action": "FOLLOW_MACRO", "macro_age_sec": 0.0, "micro_quality": 0.0, "conflict": False}
    """
    方向时效冲突解决器（三态）：
    - FOLLOW_MACRO: 维持宏观方向
    - ALLOW_MICRO_PROBE: 允许微观探针（LIMIT/小仓）
    - FLIP_TO_MICRO: 当前周期切换到微观方向
    """
    side = str(signal.get("side", "") or "").upper()
    if side not in {"BUY", "SELL"} or agent_dir not in {"LONG", "SHORT"}:
        return {"action": "FOLLOW_MACRO", "macro_age_sec": 0.0, "micro_quality": 0.0, "conflict": False}

    conflict = (agent_dir == "LONG" and side == "SELL") or (agent_dir == "SHORT" and side == "BUY")
    if not conflict:
        return {"action": "FOLLOW_MACRO", "macro_age_sec": 0.0, "micro_quality": 0.0, "conflict": False}
    regime_u = str(signal.get("regime", "UNKNOWN") or "UNKNOWN").upper()
    two_way_allowed = _is_bidirectional_regime(regime_u)

    now_ts = time.time()
    ttl = float(_get_effective_param("macro_direction_stale_sec", 7200) or 7200)
    macro_updated_ts = float((direction_data or {}).get("updated_ts", 0.0) or 0.0)
    macro_age_sec = max(0.0, now_ts - macro_updated_ts) if macro_updated_ts > 0 else 9999.0
    macro_phase = str((direction_data or {}).get("phase", "UNKNOWN") or "UNKNOWN").upper()
    macro_prob = float((direction_data or {}).get("probability", 0.5) or 0.5)

    event_tag = str(indicators.get("event", "") or "").upper()
    adx_v = float(indicators.get("adx", 0.0) or 0.0)
    ema9_v = float(indicators.get("ema9", 0.0) or 0.0)
    ema21_v = float(indicators.get("ema21", 0.0) or 0.0)
    conf = float(signal.get("confidence", 0.0) or 0.0)
    ema_align = (side == "BUY" and ema9_v > ema21_v) or (side == "SELL" and ema9_v < ema21_v)
    bos_align = (side == "BUY" and event_tag == "BOS_UP") or (side == "SELL" and event_tag == "BOS_DOWN")
    micro_quality = 0.0
    micro_quality += 0.35 if bos_align else 0.0
    micro_quality += 0.25 if ema_align else 0.0
    micro_quality += min(0.25, max(0.0, (adx_v - 20.0) / 40.0))
    micro_quality += min(0.15, max(0.0, (conf - 0.55)))
    micro_quality = max(0.0, min(1.0, micro_quality))

    resolver_state = state.get("direction_resolver", {}) if isinstance(state.get("direction_resolver", {}), dict) else {}
    last_side = str(resolver_state.get("last_conflict_side", "") or "")
    last_ts = float(resolver_state.get("last_conflict_ts", 0.0) or 0.0)
    streak = int(resolver_state.get("conflict_streak", 0) or 0)
    if side == last_side and (now_ts - last_ts) <= max(120.0, ttl * 4):
        streak += 1
    else:
        streak = 1

    macro_degraded = bool((direction_data or {}).get("degraded", False))
    macro_stale = macro_degraded or macro_age_sec > max(ttl * 1.35, 90.0) or macro_phase in {"LATE", "DEGRADED", "FAILED"}

    # ---- 三维宏观权威度 macro_authority = τ(时间) × δ(价格验证) × σ(波动率) ----
    macro_age_h = macro_age_sec / 3600.0
    tau = max(0.05, 1.0 - (macro_age_h / 6.0) ** 0.8)  # 前2h缓慢衰减，后加速

    dir_indicators = (direction_data or {}).get("indicators", {}) or {}
    if not isinstance(dir_indicators, dict):
        dir_indicators = {}
    dir_ref_price = float(dir_indicators.get("bid", 0.0) or 0.0)
    current_price = float(indicators.get("bid", 0.0) or 0.0)
    atr_v = max(float(indicators.get("atr", 1.0) or 1.0), 0.1)

    if agent_dir == "LONG" and dir_ref_price > 0 and current_price > 0:
        drift_atr = (dir_ref_price - current_price) / atr_v  # >0=反向下跌
    elif agent_dir == "SHORT" and dir_ref_price > 0 and current_price > 0:
        drift_atr = (current_price - dir_ref_price) / atr_v  # >0=反向上涨
    else:
        drift_atr = 0.0

    if drift_atr >= 0:
        delta = max(0.30, 1.0 - drift_atr * 0.25)          # 背离惩罚
    else:
        delta = min(1.05, 1.0 + abs(drift_atr) * 0.03)     # 顺向微强化(≤5%)

    atr_pct = min(5.0, atr_v / max(current_price, 1.0) * 100.0) if current_price > 0 else 0.15
    sigma = max(0.60, min(1.20, 1.20 - (atr_pct - 0.10) * 2.0))

    macro_authority = max(0.0, min(1.0, tau * delta * sigma))

    # ---- 权威度决策矩阵 ----
    ma = config_manager.get_macro_authority_thresholds()
    if macro_authority >= ma["high"]:
        probe_th = 0.75
        flip_th  = 0.90
    elif macro_authority >= ma["medium"]:
        probe_th = 0.62
        flip_th  = 0.80
    elif macro_authority >= ma["low"]:
        probe_th = 0.50
        flip_th  = 0.70
    else:
        probe_th = 0.40
        flip_th  = 0.60

    if not two_way_allowed:
        probe_th = min(1.0, probe_th + 0.08)
        flip_th  = min(1.0, flip_th  + 0.08)

    # 冲突持续 N 周期 → 降阈值，让微信号更容易通过（持续相悖说明宏观可能已过时）
    if streak >= 3:
        probe_th = max(0.30, probe_th - 0.08)
        flip_th  = max(0.45, flip_th  - 0.08)
    elif streak >= 2:
        probe_th = max(0.35, probe_th - 0.04)
        flip_th  = max(0.50, flip_th  - 0.04)

    action = "FOLLOW_MACRO"
    if micro_quality >= flip_th:
        action = "FLIP_TO_MICRO"
    elif micro_quality >= probe_th:
        action = "ALLOW_MICRO_PROBE"

    state["direction_resolver"] = {
        "last_conflict_side": side,
        "last_conflict_ts": now_ts,
        "conflict_streak": streak,
        "macro_age_sec": round(macro_age_sec, 1),
        "macro_phase": macro_phase,
        "macro_prob": round(macro_prob, 3),
        "macro_degraded": bool(macro_degraded),
        "macro_authority": round(macro_authority, 4),
        "micro_quality": round(micro_quality, 3),
        "action": action,
        "two_way_allowed": bool(two_way_allowed),
        "regime": regime_u,
    }
    return {
        "action": action,
        "macro_age_sec": macro_age_sec,
        "macro_authority": round(macro_authority, 4),
        "micro_quality": micro_quality,
        "conflict": True,
        "streak": streak,
        "macro_stale": macro_stale,
        "macro_prob": macro_prob,
        "macro_degraded": bool(macro_degraded),
        "two_way_allowed": bool(two_way_allowed),
        "regime": regime_u,
    }

def save_agent_direction(direction, reason, confidence, indicators=None, probability=None, phase=None, degraded=False, metadata=None, conviction=None, timeframe=None, risk_appetite=None, key_levels=None, macro_regime=None):
    """保存 Agent 的方向决策（委托 direction_registry 核心写入 + 本地 side effects）"""
    direction = str(direction or "LONG").upper()
    data = _registry_save_agent_direction(
        direction, reason, confidence,
        indicators=indicators, probability=probability, phase=phase,
        degraded=degraded, metadata=metadata,
        conviction=conviction, timeframe=timeframe,
        risk_appetite=risk_appetite, key_levels=key_levels,
        macro_regime=macro_regime,
    )
    normalized_direction = data.get("direction", "NEUTRAL")
    _daemon_log(f"Agent direction SAVED: {normalized_direction} ({reason})")

    try:
        snapshot = _get_runtime_blackboard().get_snapshot(include_logs=0)
        macro_state = snapshot.get("Macro_Commander", {}) if isinstance(snapshot, dict) else {}
        current_calls = int(macro_state.get("daily_calls", 0) or 0)
        _runtime_blackboard_update("Macro_Commander", "bias", normalized_direction)
        _runtime_blackboard_update("Macro_Commander", "confidence", float(confidence))
        _runtime_blackboard_update("Macro_Commander", "status", "DONE")
        _runtime_blackboard_update("Macro_Commander", "degraded", bool(degraded))
        _runtime_blackboard_update("Macro_Commander", "daily_calls", current_calls + 1)
    except Exception as e2:
        _daemon_log(f"sync Macro_Commander failed: {e2}")

    try:
        sys.path.insert(0, str(ROOT))
        from skill_trading_memory import append_agent_decision
        append_agent_decision(
            direction=normalized_direction,
            reason=reason,
            confidence=confidence,
            indicators=indicators or {},
        )
    except Exception as e2:
        _daemon_log(f"append_agent_decision failed: {e2}")

    # ── 同步写入 agent_state.json（供 direction_registry / decision_snapshot 读取）──
    try:
        _agent_state_path = config_manager.get_data_dir() / "bridge" / "state" / "agent_state.json"
        if _agent_state_path.exists():
            _state = json.loads(_agent_state_path.read_text(encoding="utf-8"))
        else:
            _state = {}
        if isinstance(_state, dict):
            _state["agent_direction"] = {
                "direction": normalized_direction,
                "confidence": float(confidence or 0.5),
                "probability": float(probability or confidence or 0.5),
                "reason": str(reason or ""),
                "phase": str(phase or "FRESH").upper(),
                "degraded": bool(degraded),
                "updated": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
                "updated_ts": time.time(),
                "valid_until": time.time() + DEFAULT_MACRO_TTL_SEC,
                "source": "macro_llm_direction",
            }
            _agent_state_path.parent.mkdir(parents=True, exist_ok=True)
            _agent_state_path.write_text(json.dumps(_state, ensure_ascii=False, indent=2), encoding="utf-8")
    except Exception as _e3:
        _daemon_log(f"sync agent_state agent_direction failed: {_e3}")

    return data

def should_reverse_direction(old_dir, new_dir):
    """检测是否需要反向（重大方向转换）"""
    opposites = {"LONG": "SHORT", "SHORT": "LONG"}
    if old_dir in opposites and opposites[old_dir] == new_dir:
        return True  # 多转空或空转多
    return False

def _agent_direction_to_side(agent_dir):
    normalized = str(agent_dir or "").upper()
    if normalized == "LONG":
        return "BUY"
    if normalized == "SHORT":
        return "SELL"
    return None

# ============ Governor ============
def governor_check(price_bid, price_ask, state):
    spread = round((price_ask - price_bid) * 100, 1)

    # 读取参数建议（每次检查刷新缓存）
    _load_params_suggestion()

    # 获取当前生效参数
    spread_kill = float(_get_effective_param("spread_kill", 70))
    kill_switch = state.get("kill_switch", False)
    total_loss_snapshot = _get_total_loss_stop_snapshot(state)
    
    max_consec = int(_get_effective_param("max_consecutive_losses", 4))

    # Governor 实时拦截逻辑（不修改底层配置）
    if spread > spread_kill: return False, f"WIDE_SPREAD_{spread}"
    if bool(total_loss_snapshot.get("active", False)):
        current_pct = float(total_loss_snapshot.get("current_pct", 0.0) or 0.0)
        threshold_pct = float(total_loss_snapshot.get("threshold_pct", 0.0) or 0.0)
        source = str(total_loss_snapshot.get("current_source", "NONE") or "NONE")
        return False, f"TOTAL_LOSS_STOP_{current_pct:.2f}%>={threshold_pct:.2f}%[{source}]"
    if state.get("consecutive_losses", 0) >= max_consec: return False, "LOSS_LIMIT"
    return True, "APPROVED"

# ============ [DUAL-CALL] RiskManager 对比验证 =============
# 阶段二过渡期：新旧风控同时运行，对比结果，确保 RiskManager 行为与旧逻辑一致
# 确认无误后，可切换到纯 RiskManager，删除旧 governor_check

_RISK_MGR = None  # 延迟加载

def _get_risk_manager():
    """延迟加载 RiskManager（避免循环导入）"""
    global _RISK_MGR
    if _RISK_MGR is None:
        try:
            from risk_manager import GoldRiskManager
            _RISK_MGR = GoldRiskManager()
            _daemon_log("RiskManager loaded successfully")
        except Exception as e:
            _daemon_log(f"RiskManager load failed: {e}")
    return _RISK_MGR


def _compare_risk_results(old_approved: bool, old_reason: str,
                           new_result: dict, signal: dict) -> None:
    """
    对比新旧风控结果，记录差异（用于验证阶段）
    如果新结果与旧结果不一致，打印告警日志
    """
    new_action = new_result.get("payload", {}).get("action", "HOLD")
    new_drop = new_result.get("payload", {}).get("drop_reason")

    # 判定旧系统是否通过
    old_pass = old_approved  # True = APPROVED, False = BLOCKED

    # 判定新系统是否通过
    new_pass = (new_action != "HOLD")

    if old_pass != new_pass:
        _daemon_log(
            f"[RISK MISMATCH] old={'APPROVED' if old_pass else 'BLOCKED:'+old_reason} | "
            f"new={'APPROVED' if new_pass else 'BLOCKED:'+str(new_drop)} | "
            f"signal={signal.get('action')} conf={signal.get('confidence')} | "
            f"trace_id={new_result.get('trace_id')}"
        )


def governor_check_with_rm(price_bid, price_ask, state, signal=None):
    """
    双调风控：同时调用旧 governor_check 和新 RiskManager
    返回：(approved, reason, risk_decision)
    - approved/reason: 旧逻辑结果（保持向后兼容）
    - risk_decision: 新 RiskManager 结果（用于后续流程）
    """
    # 1. 旧逻辑（保持原有行为）
    approved, reason = governor_check(price_bid, price_ask, state)

    # 2. 新逻辑（RiskManager，阶段二验证用）
    rm = _get_risk_manager()
    risk_decision = None

    if rm is not None and signal is not None:
        try:
            # 构造 context
            spread = round((price_ask - price_bid) * 100, 1)
            context = {
                "price_bid": price_bid,
                "price_ask": price_ask,
                "spread": spread,
                "state": state,
                "params": {}
            }

            # 包装 signal（兼容 RiskManager 的 payload 格式）
            wrapped = {
                "schema_version": "2.1.0",
                "trace_id": f"daemon_{int(time.time()*1000)}",
                "strategy_id": "ema_macd_v2",
                "payload": {
                    "action": signal.get("action", "HOLD"),
                    "confidence": signal.get("confidence", 0),
                    "side": signal.get("side", "BUY"),
                    "entry": signal.get("entry", 0),
                    "sl": signal.get("sl", 0),
                    "tp": signal.get("tp", 0),
                    "indicators": signal.get("indicators", {}),
                }
            }

            risk_decision = rm.evaluate(wrapped, context)

            # RM veto: if new RiskManager blocks, override old governor
            if risk_decision is not None:
                rm_payload = risk_decision.get("payload", {}) if isinstance(risk_decision, dict) else {}
                rm_action = str(rm_payload.get("action", "APPROVED") or "APPROVED").upper()
                rm_drop = str(rm_payload.get("drop_reason", "") or "").strip()
                if rm_action == "HOLD" and rm_drop:
                    approved = False
                    reason = f"RM_VETO:{rm_drop}"
                    _daemon_log(
                        f"[RM_VETO] New RiskManager overruled old governor: "
                        f"old={reason} new_drop={rm_drop}",
                        "WARN",
                    )

            # 对比结果（只在有信号时对比）
            if signal.get("action") != "HOLD":
                _compare_risk_results(approved, reason, risk_decision, signal)

        except Exception as e:
            _daemon_log(f"RiskManager evaluate error: {e}")

    # 3. 返回（新 RM 可否决旧 governor）
    return approved, reason, risk_decision


# ============ Agent 复核系统 ============
def write_review_request(signal_data, reason):
    """写入复核请求，已有同内容待复核则跳过"""
    REVIEW_DIR.mkdir(parents=True, exist_ok=True)

    # ── 僵尸复核文件自动清理（价格漂移 + 超时）──────────────
    # 获取当前价格用于漂移判断
    try:
        cur_bid, cur_ask, _, _, _ = get_price()
    except Exception:
        cur_bid = cur_ask = 0

    try:
        for rf in list(REVIEW_DIR.glob("*.review.json")):
            try:
                data = json.loads(rf.read_text(encoding="utf-8"))
                # 超时清理：底层是 5 分钟 K 线模型，超过 3 根 K 线（15分钟）直接视为失效
                ts = data.get("timestamp", "")
                if ts:
                    try:
                        file_age = time.time() - datetime.fromisoformat(ts.replace("Z","+00:00")).timestamp()
                        if file_age > 900:  # 15分钟 (900秒)
                            rf.unlink(missing_ok=True)
                            _daemon_log(f"[GC] Removed stale review (age={file_age/60:.1f}m > 15m): {rf.name}")
                            continue
                    except Exception:
                        pass
                # 漂移清理：当前价格已偏离 entry，因为是短线模型，超过 1.5×ATR 即视为错过入场点
                sig_entry = data.get("entry", 0)
                sig_atr = data.get("indicators", {}).get("atr14", 0) or 0
                if sig_entry > 0 and sig_atr > 0 and cur_bid > 0:
                    drift = abs(cur_bid - sig_entry)
                    if drift > sig_atr * 1.5:
                        rf.unlink(missing_ok=True)
                        _daemon_log(f"[GC] Removed drifted review: {rf.name} drift={drift:.1f} ({sig_atr*1.5:.1f}ATR threshold)")
            except Exception:
                pass
    except Exception:
        pass

    # 根据 signal 内容生成稳定 ID（同一信号不重复创建 review）
    entry = signal_data.get("entry", 0)
    sl = signal_data.get("sl", 0)
    tp = signal_data.get("tp", 0)
    side = signal_data.get("side", "?")

    # ── TP/SL 比例自动校验（关键防线）──────────────────────────
    # 修复 Bug 1：方向性验证，确保止损止盈在正确的物理位置
    if side == "BUY" and (sl >= entry or tp <= entry):
        _log(f"[AUDIT] [FAIL] 价格倒挂！BUY entry={entry}, sl={sl} (应<entry), tp={tp} (应>entry)，拒绝写入复核", "ERROR")
        return None
    if side == "SELL" and (sl <= entry or tp >= entry):
        _log(f"[AUDIT] [FAIL] 价格倒挂！SELL entry={entry}, sl={sl} (应>entry), tp={tp} (应<entry)，拒绝写入复核", "ERROR")
        return None

    atr = signal_data.get("indicators", {}).get("atr14", 0) or 0
    if atr > 0:
        sl_dist = abs(entry - sl)
        tp_dist = abs(tp - entry)
        rr = tp_dist / max(sl_dist, 0.001)
        if rr < 1.0:
            _log(f"[AUDIT] [WARN]️ TP/SL盈亏比{rr:.2f}<1.0偏低（TP={tp_dist:.1f}点 SL={sl_dist:.1f}点），标记低RR", "WARN")
    # 用 entry+sl+tp 组合做 ID，同一信号不会重复创建
    stable_id = f"review_{side}_{round(entry,1)}_{round(sl,1)}_{round(tp,1)}"

    rf = REVIEW_DIR / f"{stable_id}.review.json"
    # 已有且未处理的 review 不重复写入
    if rf.exists():
        existing = {}
        try: existing = json.loads(rf.read_text(encoding="utf-8"))
        except: pass
        if existing.get("status") == "PENDING":
            _daemon_log(f"Review already pending: {stable_id}, skipping duplicate")
            return stable_id  # 已有待复核，跳过

    sig_id = f"{stable_id}_{int(time.time())}"  # 加时间戳防冲突
    rf.write_text(json.dumps({
        "signal_id": sig_id,
        "stable_id": stable_id,
        "decision_id": signal_data.get("_learning_decision_id", signal_data.get("signal_id", sig_id)),
        "action": signal_data.get("action"),
        "side": side,
        "entry": entry,
        "sl": sl,
        "tp": tp,
        "confidence": signal_data.get("confidence", 0),
        "regime": signal_data.get("regime", "?"),
        "reason": reason,
        "indicators": signal_data.get("indicators", {}),
        "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ"),
        "status": "PENDING",
    }, indent=2))
    _daemon_log(f"Review request: {sig_id} reason={reason}")
    return sig_id

def check_review_approved_any():
    """
    检查是否有任何待处理的 approve/reject 文件。
    返回: ("APPROVED", sig_id, stable_id) / ("REJECTED", sig_id, stable_id) / (None, None, None)
    """
    # ── 增加僵尸文件自动清理 (Garbage Collection) ──
    now = time.time()
    try:
        if REVIEW_DIR.exists():
            for pending_f in REVIEW_DIR.glob("*.review.json"):
                # 5分钟模型信号最长只保留 15 分钟
                if now - pending_f.stat().st_mtime > 900:
                    _daemon_log(f"[CLEAN] 清理过期僵尸复核文件: {pending_f.name}")
                    pending_f.unlink(missing_ok=True)
    except Exception as e:
        _daemon_log(f"清理僵尸复核文件出错: {e}")

    for d in [REVIEW_DIR / "approve", REVIEW_DIR / "reject"]:
        if not d.exists(): continue
        files = sorted(d.glob("*.json"), key=lambda x: x.stat().st_mtime, reverse=True)
        if files:
            f = files[0]
            try:
                data = json.loads(f.read_text(encoding="utf-8"))
                decision = "APPROVED" if "approve" in str(d) else "REJECTED"
                sig_id = data.get("signal_id", f.name)
                stable_id = data.get("stable_id", sig_id)
                f.unlink(missing_ok=True)
                return decision, sig_id, stable_id
            except:
                pass
    return None, None, None

def process_review_result(decision, sig_id, stable_id=None):
    """处理 approve/reject 结果"""
    if decision == "APPROVED":
        # 找到对应的 review 文件（用 stable_id 定位）
        try:
            rf = REVIEW_DIR / f"{stable_id or sig_id}.review.json"
            if rf.exists():
                sig = json.loads(rf.read_text(encoding="utf-8"))
                _block_legacy_review_open(sig, sig_id=sig_id, stable_id=stable_id)
                rf.write_text(json.dumps({**sig, "status": "BLOCKED_LEGACY_REVIEW_OPEN_DISABLED"}, indent=2))
                return
        except Exception as e:
            _daemon_log(f"process_review_result error: {e}")
    elif decision == "REJECTED":
        try:
            rf = REVIEW_DIR / f"{stable_id or sig_id}.review.json"
            if rf.exists():
                sig = json.loads(rf.read_text(encoding="utf-8"))
                decision_id = sig.get("decision_id", stable_id or sig_id)
                _close_decision_state(
                    decision_id,
                    "VETOED",
                    reason="REVIEW_REJECTED",
                    review_signal_id=sig.get("signal_id", sig_id),
                    stable_id=sig.get("stable_id", stable_id or sig_id),
                    review_reason=sig.get("reason", ""),
                )
                rf.write_text(json.dumps({**sig, "status": "REJECTED_BY_REVIEW"}, indent=2))
        except Exception as e:
            _daemon_log(f"process_review_result reject error: {e}")
        _log(f"[FAIL] Review否决: sig_id={sig_id}", "WARN")

def check_review_result(sig_id):
    for d in [REVIEW_DIR / "approve", REVIEW_DIR / "reject"]:
        if (d / f"{sig_id}.json").exists(): return "APPROVED" if "approve" in str(d) else "REJECTED"
    return "PENDING"

def wait_for_agent_review(sig_id, timeout=120): # 修改超时时间为 120 秒，对齐 Agent 的修复
    waited = 0
    while waited < timeout:
        time.sleep(10); waited += 10
        result = check_review_result(sig_id)
        if result in ("APPROVED", "REJECTED"):
            _daemon_log(f"Review done: {sig_id} => {result}")
            return result
    _daemon_log(f"Review timeout: {sig_id}")
    return "TIMEOUT"

# ============ 策略模块（只分析，不决策方向）============
def generate_signal(price_bid, price_ask, state):
    """[DEPRECATED] Pipeline generates signals via _build_instance_signals() + run_engine().

    This legacy gold_trader signal generation path is retired. All signal generation
    is managed by pipeline.py's multi-strategy regime-directed routing.
    """
    _daemon_log("[DEPRECATED] generate_signal called — pipeline.py manages all signal generation.", "WARN")
    return {
        "action": "HOLD", "regime": "TRANSITION", "selection_regime": "TRANSITION",
        "confidence": 0, "reason": "RETIRED_PIPELINE_MANAGES_SIGNAL_GENERATION",
        "indicators": {}, "session": {},
        "volume_lots": 0.01,
    }


def _validate_params(suggestions):
    """
    验证外部参数建议列表，返回通过/拒绝分类。
    
    验证逻辑（拒绝优于钳制）：
    1. 参数必须存在于 PARAM_SCHEMA 中，否则拒绝（未知参数不可接受）
    2. 值类型必须与 Schema 一致，否则拒绝（不接受隐式转换）
    3. 值必须在 [min, max] 范围内，否则拒绝（不接受静默钳制）
    4. 通过全部检查的参数，以 (key, typed_value) 形式返回
    
    Returns:
        (validated: dict, rejected: list[dict], warnings: list[str])
    """
    validated = {}
    rejected = []
    warnings = []
    if "_PARAM_MAP" not in globals() or "PARAM_SCHEMA" not in globals():
        warnings.append("legacy_param_schema_retired")
        return validated, rejected, warnings

    for s in suggestions:
        param = s.get("param")
        raw_val = s.get("applied_value")
        source = s.get("id", "?")
        note = s.get("note", "")

        # ── 检查1：参数是否在 Schema 中 ──
        if param not in _PARAM_MAP:
            rejected.append({
                "param": param, "value": raw_val, "source": source,
                "reason": f"未知参数（不在 PARAM_SCHEMA 中），拒绝加载"
            })
            continue

        gen_key = _PARAM_MAP[param]
        if gen_key not in PARAM_SCHEMA:
            rejected.append({
                "param": param, "value": raw_val, "source": source,
                "reason": f"参数映射到 '{gen_key}' 但该参数不在 Schema 中，拒绝加载"
            })
            continue

        schema = PARAM_SCHEMA[gen_key]
        expected_type = schema["type"]
        lo, hi = schema["min"], schema["max"]
        default = schema["default"]

        # ── 检查2：类型是否正确 ──
        type_ok = False
        typed_val = None
        if expected_type == int:
            if isinstance(raw_val, int) and not isinstance(raw_val, bool):
                typed_val = raw_val
                type_ok = True
            elif isinstance(raw_val, float) and raw_val == int(raw_val):
                typed_val = int(raw_val)
                warnings.append(f"[{gen_key}] 类型警告: float({raw_val})→int({typed_val})，建议修复数据源")
                type_ok = True
            elif isinstance(raw_val, str) and raw_val.isdigit():
                typed_val = int(raw_val)
                type_ok = True
        elif expected_type == float:
            if isinstance(raw_val, (int, float)) and not isinstance(raw_val, bool):
                typed_val = float(raw_val)
                type_ok = True
            elif isinstance(raw_val, str):
                try:
                    typed_val = float(raw_val)
                    type_ok = True
                except ValueError:
                    pass

        if not type_ok:
            rejected.append({
                "param": param, "value": raw_val, "source": source,
                "reason": f"类型错误: 期望 {expected_type.__name__}，实际 {type(raw_val).__name__}，拒绝加载"
            })
            continue

        # ── 检查3：值是否在允许范围内 ──
        if not (lo <= typed_val <= hi):
            rejected.append({
                "param": param, "value": raw_val, "source": source,
                "reason": f"值 {typed_val} 超出安全范围 [{lo}, {hi}]，拒绝加载（将使用默认值 {default}）"
            })
            continue

        validated[gen_key] = typed_val

    return validated, rejected, warnings


def _apply_effective_params_to_generator(gen):
    """
    从 effective_params.json 读取参数建议，Schema 验证后应用到信号生成器。
    所有外部参数必须经过 PARAM_SCHEMA 验证——验证失败 = 拒绝，不做钳制。
    """
    try:
        if "_PARAM_MAP" not in globals() or "PARAM_SCHEMA" not in globals():
            _daemon_log(
                "[DEPRECATED] legacy PARAM_SCHEMA retired; skip effective_params application",
                "WARN",
            )
            return
        eff_file = LOGS_DIR / "effective_params.json"
        if not eff_file.exists():
            _daemon_log("effective_params.json 不存在，使用生成器默认值")
            return

        import json
        data = json.loads(eff_file.read_text(encoding="utf-8"))
        suggestions = data.get("suggestions", [])
        if not suggestions:
            _daemon_log("effective_params.json 为空，使用生成器默认值")
            return

        # ── 第一步：Schema 验证（拒绝优于钳制）──
        validated, rejected, warnings = _validate_params(suggestions)

        # 记录所有被拒绝的参数（这些参数不会被应用）
        for r in rejected:
            _daemon_log(
                f"[SCHEMA [FAIL] REJECTED] param={r['param']} value={r['value']} "
                f"source={r['source']} | reason={r['reason']}"
            )

        for w in warnings:
            _daemon_log(f"[SCHEMA [WARN]️ WARNING] {w}")

        # ── 第二步：仅应用通过验证的参数 ──
        applied = []
        for gen_key, val in validated.items():
            old_val = gen.params.get(gen_key)
            gen.params[gen_key] = val
            applied.append(f"{gen_key}={old_val}→{val}")

        if applied:
            _daemon_log(f"[SCHEMA [OK] APPLIED] {', '.join(applied)}")
        else:
            _daemon_log("[SCHEMA] 所有外部参数均被拒绝，使用生成器默认值")

    except Exception as e:
        _daemon_log(f"_apply_effective_params_to_generator failed: {e}")


def _effective_strategy_payload() -> dict:
    eff_file = LOGS_DIR / "effective_params.json"
    if not eff_file.exists():
        return {}
    try:
        import json as _json

        return _json.loads(eff_file.read_text(encoding="utf-8"))
    except Exception:
        return {}


def _effective_strategy_type(default: str = "ema_macd") -> str:
    # [DEPRECATED] Manual strategy override is retired — pipeline manages strategy selection.
    # _MANUAL_STRATEGY_OVERRIDE is permanently None.
    return str(default or "ema_macd")


def set_strategy_override(strategy_type: Optional[str] = None) -> Optional[str]:
    # [DEPRECATED] Strategy override is retired. Pipeline's regime-directed multi-strategy
    # routing manages all strategy selection. GUI should use unified_strategy_state APIs.
    _daemon_log("[DEPRECATED] set_strategy_override called — strategy override is retired. "
                "Strategy selection is managed by pipeline's regime-directed routing.", "WARN")
    return None


def _runtime_identity_from_generator(gen) -> dict:
    runtime_identity = {}
    try:
        identity = getattr(gen, "get_runtime_identity", None)
        if callable(identity):
            runtime_identity = identity() or {}
    except Exception:
        runtime_identity = {}
    return runtime_identity if isinstance(runtime_identity, dict) else {}

def _get_signal_generator():
    """[DEPRECATED] Legacy signal generator loader — pipeline manages all signal generation.

    This function is retained only for backward compatibility with trader_api/gui_backend
    callers that pre-warm a generator. It returns a stub that always produces HOLD signals.
    The real signal generation is handled by pipeline.py via run_engine().
    """
    _daemon_log("[DEPRECATED] _get_signal_generator called — pipeline.py manages all signal generation. "
                "The legacy gold_trader signal generator path is retired.", "WARN")
    # Return a lightweight stub that always produces HOLD
    global _SIGNAL_GEN
    if _SIGNAL_GEN is None:
        try:
            _SIGNAL_GEN = get_strategy_runtime().get_legacy_compatible_generator("ema_macd")
        except Exception:
            _SIGNAL_GEN = None
    return _SIGNAL_GEN


def reset_signal_generator():
    """[DEPRECATED] Legacy reset — pipeline manages strategy lifecycle independently."""
    global _SIGNAL_GEN
    _daemon_log("[DEPRECATED] reset_signal_generator called — pipeline manages all strategy lifecycle. "
                "Use unified_strategy_state APIs for strategy management.", "WARN")
    _SIGNAL_GEN = None
    return "retired"


def get_signal_generator_status() -> dict:
    """[DEPRECATED] Legacy status query — pipeline manages signal generation.

    Returns a retired status. GUI should query pipeline's runtime scheduler status instead.
    """
    return {
        "status": "retired",
        "reason": "pipeline_manages_all_signal_generation",
        "strategy_type": None,
        "strategy_name": None,
        "strategy_id": None,
        "strategy_version": None,
    }

# ═══════════════════════════════════════════════════════════════════════════
# [DEPRECATED] Legacy regime-strategy mapping — SUPERSEDED by:
#   src/strategy_registry.py::CANONICAL_STRATEGY_REGISTRY
#   src/strategy_registry.py::is_strategy_regime_compatible()
#   src/strategy_registry.py::list_compatible_strategies()
#
# Pipeline's regime_router.py (4D classifier) + strategy_registry provide the
# single source of truth for regime → strategy selection. These legacy maps
# are retained ONLY for gold_trader.py internal backward compatibility and
# should NOT be used by any new code.
# ═══════════════════════════════════════════════════════════════════════════
REGIME_STRATEGY_MAP = {}       # Deprecated — empty, no mappings
SUBTYPE_STRATEGY_MAP = {}      # Deprecated — empty, no mappings
REGIME_BLACKLIST = {}          # Deprecated — empty, no blacklist

# 缓存策略实例（避免每次重建）
_STRATEGY_CACHE: dict = {}
_AUTO_SELECT_ENABLED = False   # [DEPRECATED] Permanently disabled — pipeline manages selection
_MANUAL_STRATEGY_OVERRIDE: Optional[str] = None  # Permanently None


def _detect_regime_for_selection(bars: list) -> str:
    """[DEPRECATED] Legacy ADX-based regime detector — pipeline uses 4D classifier instead.

    Returns 'TRANSITION' always. Pipeline's regime_router.classify_regime() is the
    single source of truth for regime classification.
    """
    return "TRANSITION"


def _get_best_strategy_for_regime(regime: str) -> list[str]:
    """[DEPRECATED] Pipeline uses list_compatible_strategies() from strategy_registry."""
    _daemon_log("[DEPRECATED] _get_best_strategy_for_regime called — "
                "pipeline uses list_compatible_strategies() from src/strategy_registry instead.", "WARN")
    return ["ema_macd"]


def _resolve_strategy_with_structure(regime: str, default_strategy: str, structure: dict) -> tuple[str, str]:
    """[DEPRECATED] Pipeline uses strategy_registry + regime_router for structure-aware selection."""
    return default_strategy, "retired-pipeline-manages-selection"


def _build_fallback_candidates(primary: str, structure: dict) -> list[str]:
    """[DEPRECATED] Pipeline uses compatible_candidate_strategy_ids from _build_strategy_route."""
    return []


def _get_cached_generator(strategy_type: str):
    """获取或创建策略实例（带缓存）"""
    if strategy_type not in _STRATEGY_CACHE:
        try:
            _STRATEGY_CACHE.clear()  # 每次只缓存一个
            _STRATEGY_CACHE[strategy_type] = get_strategy_runtime().get_legacy_compatible_generator(
                strategy_type,
                force_reload=True,
            )
            _apply_effective_params_to_generator(_STRATEGY_CACHE[strategy_type])
            _daemon_log(f"[AutoStrategy] 市场驱动 → 策略切换: {strategy_type}")
        except Exception as e:
            _daemon_log(f"[AutoStrategy] 策略创建失败 {strategy_type}: {e}")
            return None
    return _STRATEGY_CACHE[strategy_type]


# ============ EA 响应处理 ============
def process_ea_reports(state):
    def _parse_est_profit(text: str):
        if not text:
            return None
        m = re.search(r"ALL_CLOSED_EST_PROFIT_(-?\d+(?:\.\d+)?)", text)
        if m:
            try:
                return float(m.group(1))
            except Exception:
                return None
        m = re.search(r"EST_PROFIT_(-?\d+(?:\.\d+)?)", text)
        if m:
            try:
                return float(m.group(1))
            except Exception:
                return None
        return None

    def _extract_normalized_close_pnl(report: dict):
        # 统一 CLOSE 盈亏口径：优先显式字段，其次 retcode/reason 中的 EST_PROFIT。
        explicit = []
        for key in ("pnl", "close_profit", "profit"):
            if key in report and report.get(key) is not None:
                try:
                    explicit.append((float(report.get(key)), key))
                except Exception:
                    pass
        for v, k in explicit:
            if abs(v) > 1e-9:
                return v, k
        retcode = str(report.get("retcode", "") or "")
        reason = str(report.get("reason", "") or "")
        est = _parse_est_profit(retcode) if retcode else None
        if est is None and reason:
            est = _parse_est_profit(reason)
        if est is not None:
            return float(est), "est_profit"
        if explicit:
            return explicit[0][0], explicit[0][1]
        # 兜底：用 MT5 实时价格估算（防止 PnL 记为 0 导致看板数据错误）
        try:
            from src.data_layer import get_live_bid_ask
            bid, ask = get_live_bid_ask()
            side = str(state.get("position_side", "BUY") or "BUY").upper()
            entry = float(state.get("position_entry", 0.0) or 0.0)
            vol = float(state.get("position_vol", 0.01) or 0.01)
            if entry > 0 and ((side == "BUY" and bid > 0) or (side == "SELL" and ask > 0)):
                if side == "BUY":
                    est_pnl = (bid - entry) * vol * 100.0
                else:
                    est_pnl = (entry - ask) * vol * 100.0
                return round(est_pnl, 2), "estimated_by_price"
        except Exception:
            pass
        return None, "missing"

    try:
        state = _reconcile_bridge_timeout_pending(state)
        # 清理 ticket=0 的伪 FILLED 挂起项，避免 EXECUTING 僵尸化。
        pending_map = state.get("ticket_zero_pending", {}) if isinstance(state.get("ticket_zero_pending", {}), dict) else {}
        now_ts = time.time()
        active_pending_decision = str(state.get("position_decision_id", "") or "").strip()
        active_pending_signal = str(state.get("position_signal_id", "") or "").strip()
        if pending_map:
            pending_target = ""
            if active_pending_decision and active_pending_decision in pending_map:
                pending_target = active_pending_decision
            elif len(pending_map) == 1:
                pending_target = next(iter(pending_map.keys()))
            if pending_target:
                state, rescued = _confirm_open_from_mt5(
                    state,
                    decision_id=pending_target,
                    signal_id=active_pending_signal,
                    pending_reason="TICKET_ZERO_PENDING_MT5_CONFIRMED",
                )
                if rescued:
                    pending_map = state.get("ticket_zero_pending", {}) if isinstance(state.get("ticket_zero_pending", {}), dict) else {}
        expired_decisions = []
        for decision_key, created_ts in pending_map.items():
            try:
                age = now_ts - float(created_ts or 0.0)
            except Exception:
                age = TICKET_ZERO_PENDING_TIMEOUT_SEC + 1
            if age >= TICKET_ZERO_PENDING_TIMEOUT_SEC:
                state, rescued = _confirm_open_from_mt5(
                    state,
                    decision_id=str(decision_key),
                    signal_id=active_pending_signal,
                    pending_reason="TICKET_ZERO_TIMEOUT_MT5_CONFIRMED",
                )
                if rescued:
                    pending_map = state.get("ticket_zero_pending", {}) if isinstance(state.get("ticket_zero_pending", {}), dict) else {}
                    continue
                expired_decisions.append(str(decision_key))
        for decision_id in expired_decisions:
            _close_decision_state(
                decision_id,
                "FAILED",
                reason="TICKET_ZERO_TIMEOUT",
                timeout_sec=TICKET_ZERO_PENDING_TIMEOUT_SEC,
            )
            pending_map.pop(decision_id, None)
        state["ticket_zero_pending"] = pending_map

        for f in sorted(_bridge_from_ea_dir().glob("*.report.json")):
            raw_text = ""
            report = None
            import re
            # Retry with backoff: EA may still be writing the file (no file locks on MQL5 side).
            # A partial read → JSON parse failure. Retry up to 3 times (0.05s / 0.15s / 0.35s)
            # before giving up and deleting the corrupted file.
            for retry_attempt in range(3):
                try:
                    raw_text = f.read_text(encoding="utf-8")
                    raw_text = re.sub(r"\"time\":\s*([^\"{,]+)\s*,", r"\"time\": \"\\1\",", raw_text)
                    report = json.loads(raw_text)
                    break
                except Exception as e:
                    if retry_attempt < 2:
                        _daemon_log(
                            f"[WARN] Report parse retry {retry_attempt + 1}/3 for {f.name}: {e}"
                        )
                        time.sleep(0.05 * (2 ** retry_attempt))
                    else:
                        _daemon_log(
                            f"[ERROR] Failed to parse report {f.name} after 3 retries: {e}. "
                            f"Raw: {raw_text[:100]}"
                        )
            if report is None:
                f.unlink(missing_ok=True)
                continue
            # 归档报告文件，防止数据丢失（方便事后审计和问题排查）
            try:
                ARCHIVE_DIR = DATA_DIR / "bridge" / "archive"
                ARCHIVE_DIR.mkdir(parents=True, exist_ok=True)
                # f.stem 是 xxx.report，再加 .report 会变成 .report.report.json
                # 直接用 f.name 就够了（不含 FROM_EA 目录的嵌套路径部分）
                archive_path = ARCHIVE_DIR / f.name
                if not archive_path.exists():
                    archive_path.write_text(json.dumps(report, ensure_ascii=False), encoding="utf-8")
                _prune_archive_dir(ARCHIVE_DIR)
            except Exception as arch_err:
                _daemon_log(f"[WARN] Failed to archive report {f.name}: {arch_err}")
            f.unlink(missing_ok=True)
            status = report.get("status", ""); sig_id = report.get("signal_id", "")
            resolved_decision_id = str(
                report.get("decision_id")
                or _resolve_decision_id(signal_id=sig_id, ticket=report.get("ticket", 0))
                or state.get("position_decision_id", "")
            ).strip()
            bridge_timeout_pending = state.get("bridge_timeout_pending", {}) if isinstance(state.get("bridge_timeout_pending", {}), dict) else {}
            if resolved_decision_id and resolved_decision_id in bridge_timeout_pending:
                bridge_timeout_pending.pop(resolved_decision_id, None)
                state["bridge_timeout_pending"] = bridge_timeout_pending
            _log(f"EA report: {status} {report.get('action','')} {sig_id} retcode={report.get('retcode','')}", "INFO")
            if status == "FILLED" and report.get("action") == "OPEN":
                # ── 防护：ticket=0 的 FILLED 是 EA 虚假确认，不代表 MT5 真实成交 ──
                # 真实成交必有非零 ticket；0 表示 EA 未收到 MT5 确认，属幽灵单
                report_ticket = report.get("ticket", 0)
                if not report_ticket:
                    _daemon_log(f"[WARN] FILLED with ticket=0 treated as PENDING (sig={sig_id}); waiting for real MT5 confirmation")
                    # 不设置 position_open，让请求维持 EXECUTING 状态，等下一轮 MT5 真实确认
                    if resolved_decision_id:
                        pending_map = state.get("ticket_zero_pending", {}) if isinstance(state.get("ticket_zero_pending", {}), dict) else {}
                        already_pending = resolved_decision_id in pending_map
                        pending_map[resolved_decision_id] = time.time()
                        state["ticket_zero_pending"] = pending_map
                        if not already_pending:
                            _transition_decision_state(resolved_decision_id, "EXECUTING", reason="TICKET_ZERO_PENDING")
                    state, rescued = _confirm_open_from_mt5(
                        state,
                        decision_id=resolved_decision_id,
                        signal_id=sig_id,
                        report=report,
                        pending_reason="TICKET_ZERO_REPORT_MT5_CONFIRMED",
                    )
                    if rescued:
                        pending_map = state.get("ticket_zero_pending", {}) if isinstance(state.get("ticket_zero_pending", {}), dict) else {}
                    continue
                if resolved_decision_id:
                    pending_map = state.get("ticket_zero_pending", {}) if isinstance(state.get("ticket_zero_pending", {}), dict) else {}
                    pending_map.pop(resolved_decision_id, None)
                    state["ticket_zero_pending"] = pending_map
                report_attribution = report.get("attribution", {}) if isinstance(report.get("attribution", {}), dict) else {}
                pending_attribution = state.get("position_attribution", {}) if isinstance(state.get("position_attribution", {}), dict) else {}
                execution_attribution = normalize_position_attribution(
                    report_attribution or pending_attribution,
                    action="OPEN",
                    fallback=pending_attribution,
                )
                pipeline_owned = _report_owned_by_pipeline(report) or str(execution_attribution.get("writer_owner", "") or "").lower() == "pipeline"
                fill_price = float(report.get("fill_price", 0) or 0)
                next_open_state = {
                    "position_open": True,
                    "position_side": report.get("side", "BUY"),
                    "position_entry": fill_price,
                    "position_price": fill_price,
                    "position_sl": float(report.get("sl_price", 0) or 0),
                    "position_tp": float(report.get("tp_price", 0) or 0),
                    "position_vol": float(report.get("volume", 0.01) or 0.01),
                    "position_ticket": int(report.get("ticket", 0) or state.get("position_ticket", 0) or 0),
                    "position_signal_id": sig_id or state.get("position_signal_id", ""),
                    "position_decision_id": resolved_decision_id or sig_id or state.get("position_decision_id", ""),
                    "position_regime": execution_attribution.get("regime", state.get("position_regime", "UNKNOWN")),
                    "position_confidence": float(report.get("confidence", state.get("position_confidence", 0.0)) or state.get("position_confidence", 0.0) or 0.0),
                    "position_time": time.time(),
                }
                next_open_state.update(position_state_fields_from_attribution(execution_attribution, action="OPEN"))
                state["l1_breakeven_done"] = False
                state["partial_done"] = False
                try:
                    mt5_state = get_mt5_position_state()
                    if mt5_state.get("position_open") and mt5_state.get("position_ticket"):
                        # 注意：position_time 使用本地时间，不从 MT5 覆盖（防止时区/重载导致时间止损误触发）
                        for k in ["position_ticket", "position_entry", "position_price", "position_sl", "position_tp", "position_vol", "position_side"]:
                            if k in mt5_state:
                                next_open_state[k] = mt5_state[k]
                except Exception as e:
                    _daemon_log(f"load mt5 position after fill failed: {e}")
                _update_account_state(mt5_state.get("balance"), mt5_state.get("equity"))

                # OPEN 状态原子一致性检查：关键信息缺失则不落地为持仓，避免半状态污染后续出场引擎。
                open_ticket = int(next_open_state.get("position_ticket", 0) or 0)
                open_entry = float(next_open_state.get("position_entry", 0) or 0)
                if open_ticket <= 0 or open_entry <= 0:
                    _daemon_log(
                        f"[OPEN_ATOMIC_ABORT] incomplete OPEN state, keep pending: "
                        f"ticket={open_ticket} entry={open_entry} sig={sig_id}"
                    )
                    if resolved_decision_id:
                        _transition_decision_state(resolved_decision_id, "EXECUTING", reason="OPEN_STATE_INCOMPLETE_PENDING")
                    continue
                state.update(next_open_state)
                state["total_trades"] += 1
                _log(f"[OK] POSITION OPENED: {state['position_side']} @{state['position_entry']} SL={state['position_sl']} TP={state['position_tp']}", "WARN")
                if not pipeline_owned:
                    try:
                        sync_open_fill(
                            execution_attribution,
                            data_dir=DATA_DIR,
                            ticket=int(state.get("position_ticket", 0) or 0),
                            side=str(state.get("position_side", "BUY") or "BUY"),
                            volume=float(state.get("position_vol", 0.0) or 0.0),
                            entry_price=float(state.get("position_entry", 0.0) or 0.0),
                            sl=float(state.get("position_sl", 0.0) or 0.0),
                            tp=float(state.get("position_tp", 0.0) or 0.0),
                            symbol="XAUUSD",
                            source="bridge_ea_filled",
                        )
                    except Exception as lifecycle_err:
                        _daemon_log(f"sync_open_fill ea report failed: {lifecycle_err}")
                try:
                    queued_post_fill_repair = _queue_post_fill_sl_tp_repair(
                        ticket=int(state.get("position_ticket", 0) or 0),
                        side=str(state.get("position_side", "") or ""),
                        fill_price=float(state.get("position_entry", 0) or 0),
                        sl=float(state.get("position_sl", 0) or 0),
                        tp=float(state.get("position_tp", 0) or 0),
                        volume=float(state.get("position_vol", 0.0) or 0.0),
                        decision_id=str(resolved_decision_id or state.get("position_decision_id", "") or ""),
                        signal_id=str(sig_id or state.get("position_signal_id", "") or ""),
                        attribution=execution_attribution,
                    )
                    if queued_post_fill_repair:
                        state["position_post_fill_rr_repair_pending"] = True
                except Exception as post_fill_err:
                    _daemon_log(f"[POST_FILL_RR_REPAIR] failed: {post_fill_err}")

                # ── Sentinel deployment: reverse conditional protection ──
                try:
                    from src.sentinel_deployer import should_deploy, deploy_sentinel
                    regime = str((execution_attribution or {}).get("regime", "UNKNOWN") or "UNKNOWN")
                    adx_v = float((execution_attribution or {}).get("adx", 25.0) or 25.0)
                    main_pos = {
                        "side": str(state.get("position_side", "BUY")),
                        "entry": float(state.get("position_entry", 0)),
                        "tp": float(state.get("position_tp", 0)),
                        "sl": float(state.get("position_sl", 0)),
                        "volume": float(state.get("position_vol", 0.01)),
                    }
                    if should_deploy(regime, adx_v, main_pos):
                        sentinel = deploy_sentinel(main_pos, float(execution_attribution.get("atr", 15.0) if isinstance(execution_attribution, dict) else 15.0))
                        if sentinel:
                            state["_sentinel_pending"] = sentinel
                            _daemon_log(f"[SENTINEL] deployed {sentinel['side']} STOP_LIMIT at {sentinel['trigger_price']} vol={sentinel['volume']}")
                except Exception:
                    pass
                # ── End sentinel ──

                # 记录开仓到 trading_ledger
                if not pipeline_owned:
                    _log_open(state)
                    _update_shadow_system(position_open=True, total_trades=state.get("total_trades", 0), total_pnl=state.get("total_pnl", 0.0))
                if resolved_decision_id and not pipeline_owned:
                    _close_decision_state(
                        resolved_decision_id,
                        "CONFIRMED",
                        reason="OPEN_FILLED",
                        ticket=state.get("position_ticket", 0),
                        signal_id=sig_id,
                        side=state.get("position_side"),
                        fill_price=state.get("position_entry", 0),
                        volume=state.get("position_vol", 0),
                    )
                    _append_decision_link(
                        signal_id=sig_id,
                        decision_id=resolved_decision_id,
                        action="OPEN",
                        ticket=state.get("position_ticket", 0),
                        source="ea_open_filled",
                    )
                # 持仓快照
                try:
                    sys.path.insert(0, str(ROOT))
                    from skill_trading_memory import save_position_snapshot
                    save_position_snapshot(positions=[{
                        "ticket": state["position_ticket"],
                        "side": state["position_side"],
                        "volume": state["position_vol"],
                        "entry": state["position_entry"],
                        "sl": state["position_sl"],
                        "tp": state["position_tp"],
                    }])
                except Exception as e:
                    _daemon_log(f"save_position_snapshot failed: {e}")
                if resolved_decision_id and not pipeline_owned:
                    _record_learning_execution(resolved_decision_id, report, state)
            elif status in ("SL_MODIFIED", "TP_MODIFIED"):
                action = report.get("action", "")
                reason = str(report.get("comment", "") or report.get("reason", ""))
                modify_telemetry = state.get("_exit_modify_telemetry", {}) if isinstance(state.get("_exit_modify_telemetry", {}), dict) else {}
                if status == "SL_MODIFIED":
                    try:
                        new_sl = float(report.get("sl_price", 0) or reason.split("sl=")[-1])
                        state["position_sl"] = new_sl
                        state["position_post_fill_rr_repair_pending"] = False
                        modify_telemetry["last_sl_modify_ts"] = int(time.time())
                        modify_telemetry["last_sl_value"] = new_sl
                        pending_sl_reason = str(modify_telemetry.pop("pending_sl_reason", "") or "")
                        modify_telemetry.pop("pending_sl_request_ts", None)
                        modify_telemetry.pop("pending_sl_value", None)
                        modify_telemetry.pop("pending_sl_signal_id", None)
                        state["_exit_modify_telemetry"] = modify_telemetry
                        entry = float(state.get("position_entry", 0) or 0)
                        side = state.get("position_side")
                        l1_confirmed = "L1_BREAKEVEN" in pending_sl_reason or "L1保本" in pending_sl_reason
                        if not pending_sl_reason and not bool(state.get("partial_done", False)):
                            l1_confirmed = (side == "BUY" and new_sl >= entry) or (side == "SELL" and new_sl <= entry)
                        if l1_confirmed:
                            state["l1_breakeven_done"] = True
                        deferred_tp_value = modify_telemetry.pop("pending_tp_after_sl_confirmed", None)
                        deferred_tp_reason = str(modify_telemetry.pop("pending_tp_after_sl_reason", "") or "")
                        deferred_tp_ticket = int(modify_telemetry.pop("pending_tp_after_sl_ticket", 0) or 0)
                        state["_exit_modify_telemetry"] = modify_telemetry
                        if deferred_tp_value is not None and float(state.get("position_tp", 0) or 0) > 0:
                            tp_sig = write_trade_request(
                                "MODIFY_TP",
                                state.get("position_side", "BUY"),
                                0,
                                0,
                                float(deferred_tp_value or 0.0),
                                vol=state.get("position_vol", 0.01),
                                sig_id=None,
                                ticket=deferred_tp_ticket or state.get("position_ticket", 0),
                                request_origin="sl_confirmed_tp_release",
                            )
                            if tp_sig:
                                modify_telemetry = state.get("_exit_modify_telemetry", {}) if isinstance(state.get("_exit_modify_telemetry", {}), dict) else {}
                                modify_telemetry["pending_tp_request_ts"] = int(time.time())
                                modify_telemetry["pending_tp_value"] = float(deferred_tp_value or 0.0)
                                modify_telemetry["pending_tp_signal_id"] = tp_sig
                                modify_telemetry["pending_tp_reason"] = deferred_tp_reason or "sl_confirmed_tp_release"
                                state["_exit_modify_telemetry"] = modify_telemetry
                    except Exception:
                        pass
                elif status == "TP_MODIFIED":
                    try:
                        new_tp = float(report.get("tp_price", 0) or reason.split("tp=")[-1])
                        state["position_tp"] = new_tp
                        state["position_post_fill_rr_repair_pending"] = False
                        modify_telemetry["last_tp_modify_ts"] = int(time.time())
                        modify_telemetry["last_tp_value"] = new_tp
                        modify_telemetry.pop("pending_tp_request_ts", None)
                        modify_telemetry.pop("pending_tp_value", None)
                        modify_telemetry.pop("pending_tp_signal_id", None)
                        modify_telemetry.pop("pending_tp_reason", None)
                        state["_exit_modify_telemetry"] = modify_telemetry
                    except Exception:
                        if "0" in reason:
                            state["position_tp"] = 0
            elif status == "FILLED" and report.get("action") in ("CLOSE", "CLOSE_ALL", "CLOSE_PARTIAL", "CLOSE_TICKET"):
                pnl, pnl_source = _extract_normalized_close_pnl(report)
                pipeline_owned = _report_owned_by_pipeline(report) or _position_owned_by_pipeline(state)
                reason_str = report.get("reason", "") or report.get("retcode", "")
                if pnl is None:
                    pnl = 0.0
                action_upper = str(report.get("action", "")).upper()
                report_ticket = int(report.get("ticket", 0) or 0)
                # 去重键：防止 EA 重复 CLOSE 回报污染连亏统计。
                close_sig = str(sig_id or "")
                dedupe_key = f"{action_upper}|{report_ticket}|{close_sig}|{round(float(pnl or 0.0), 2)}"
                close_dedupe = state.get("close_report_dedupe", [])
                if not isinstance(close_dedupe, list):
                    close_dedupe = []
                if dedupe_key in close_dedupe:
                    _daemon_log(f"[WARN] Duplicate CLOSE report ignored: {dedupe_key}")
                    continue
                close_dedupe.append(dedupe_key)
                state["close_report_dedupe"] = close_dedupe[-200:]
                # 无持仓时不再处理 CLOSE，避免重复 CLOSE 报告把 losses 累加到错误值。
                if (not bool(state.get("position_open", False))) and action_upper != "CLOSE_PARTIAL":
                    _daemon_log(
                        f"[WARN] Stale CLOSE ignored: action={action_upper}, ticket={report_ticket}, sig={close_sig}, pnl={pnl}"
                    )
                    continue

                fill_price = float(report.get("fill_price", 0) or 0)
                if fill_price < 1e-9 and "close_price" in report:
                    fill_price = float(report.get("close_price", 0) or 0)

                # 如果 EA 连 fill_price 都不传，尝试用 get_price 的当前价
                if fill_price < 1e-9:
                    try:
                        bid_t, ask_t, _, _, _ = get_price()
                        fill_price = ask_t if state.get("position_side") == "SELL" else bid_t
                    except:
                        fill_price = 0

                # 不再等待事后同步 — 用估算 PnL 确保看板数据正确
                if abs(float(pnl or 0.0)) < 1e-9 and pnl_source == "missing":
                    _daemon_log(f"[WARN] EA 未提供真实利润，使用实时价格估算 PnL。Reason: {reason_str}")

                # 计算持仓时长
                duration_min = 0
                pos_time = state.get("position_time", 0)
                if pos_time > 0:
                    duration_min = int((time.time() - pos_time) / 60)

                # 对于部分平仓，记录利润并更新冷却时间，但不清空持仓状态
                if report.get("action") == "CLOSE_PARTIAL":
                    _log(f"[OK] PARTIAL CLOSED: PnL={pnl}", "WARN")
                    state["partial_done"] = True
                    if not pipeline_owned:
                        try:
                            _log_trade(state, fill_price, fill_price, close_pnl=pnl, duration_min=duration_min)
                        except Exception as e:
                            _daemon_log(f"_log_trade (partial) failed: {e}")
                    # BUG FIX: 部分平仓也要触发冷却时间更新和历史记录
                    _seal_close_guard(
                        state,
                        closed_side=str(state.get("position_side", "") or ""),
                        closed_ticket=report_ticket,
                        close_action="CLOSE_PARTIAL",
                    )
                    _append_recent_closure(
                        state,
                        pnl,
                        duration_min=duration_min,
                        exit_price=fill_price,
                        close_action="CLOSE_PARTIAL",
                    )
                    _apply_realized_closure_stats(state, pnl, position_open=True)
                    save_state(state)
                else:
                    # 完全平仓逻辑
                    decision_id = str(state.get("position_decision_id", "") or resolved_decision_id or "").strip()
                    state_before_close = dict(state)
                    if not pipeline_owned:
                        try:
                            _log_trade(state, fill_price, fill_price, close_pnl=pnl, duration_min=duration_min)
                        except Exception as e:
                            _daemon_log(f"_log_trade (close_all) failed: {e}")
                            try:
                                bid2, ask2, _, _, _ = get_price()
                                if bid2 and ask2:
                                    _log_trade(state, bid2, ask2, close_pnl=pnl, duration_min=duration_min)
                                else:
                                    _daemon_log(f"[ERROR] _log_trade skipped: no price available for ticket={state.get('position_ticket')}")
                            except Exception as e2:
                                _daemon_log(f"get_price for log_trade failed: {e2}")
                    try:
                        _append_recent_closure(
                            state_before_close,
                            pnl,
                            duration_min=duration_min,
                            exit_price=fill_price,
                            close_action=action_upper,
                        )
                    except Exception as e:
                        _daemon_log(f"_append_recent_closure (filled close) failed: {e}")
                    state["recent_closures"] = state_before_close.get("recent_closures", [])
                    state["pattern_memory"] = state_before_close.get("pattern_memory", {})
                    state["runtime_adaptation"] = state_before_close.get("runtime_adaptation", {})
                    state["learning_adaptation_summary"] = state_before_close.get("learning_adaptation_summary", {})
                    # 只基于 recent_closures 重建，避免 archive 重复 CLOSE 回报污染连亏数据。
                    losses = _sync_consecutive_losses_from_recent(state)
                    _log(f"[OK] CLOSED: PnL={pnl}({pnl_source}) losses={losses}", "WARN")
                    # 平仓快照（空持仓）
                    try:
                        sys.path.insert(0, str(ROOT))
                        from skill_trading_memory import save_position_snapshot
                        save_position_snapshot(positions=[])
                    except Exception as e:
                        _daemon_log(f"save_position_snapshot failed: {e}")
                    # Always record closure - even if decision_id is empty, use ticket as fallback
                    if not decision_id:
                        decision_id = 'ema_legacy_' + str(report.get('ticket', state.get('position_ticket', 'unknown')))
                    close_attribution = normalize_position_attribution(
                        report.get("attribution", {}) if isinstance(report.get("attribution", {}), dict) else {},
                        action="CLOSE",
                        fallback=state_before_close.get("position_attribution", {}) if isinstance(state_before_close.get("position_attribution", {}), dict) else {},
                    )
                    if not pipeline_owned:
                        try:
                            sync_close_fill(
                                close_attribution,
                                data_dir=DATA_DIR,
                                ticket=int(report.get("ticket", state_before_close.get("position_ticket", 0)) or 0),
                                exit_price=float(fill_price or 0.0),
                                pnl=float(pnl or 0.0),
                                side=str(state_before_close.get("position_side", "") or ""),
                                reason=str(report.get("reason", "") or action_upper),
                                source="bridge_ea_close",
                            )
                        except Exception as lifecycle_err:
                            _daemon_log(f"sync_close_fill ea close failed: {lifecycle_err}")
                        _append_decision_link(
                            signal_id=sig_id or state.get("position_signal_id", ""),
                            decision_id=decision_id,
                            action=str(report.get("action", "")),
                            ticket=report.get("ticket", state.get("position_ticket", 0)),
                            source="ea_close_filled",
                        )
                        _record_learning_closure(
                            decision_id,
                            state_before_close,
                            pnl,
                            fill_price,
                            duration_min,
                            spread=0.0,
                        )
                        _close_decision_state(
                            decision_id,
                            "CONFIRMED",
                            reason="POSITION_CLOSED",
                            ticket=report.get("ticket", state.get("position_ticket", 0)),
                            realized_pnl=pnl,
                            fill_price=fill_price,
                            duration_min=duration_min,
                        )
                    # 彻底清空所有状态标记，包括 L2/L3 平仓标记
                    state.update({
                        "position_open":False,
                        "position_ticket":0,
                        "position_signal_id":"",
                        "position_decision_id":"",
                        "position_regime":"UNKNOWN",
                        "position_strategy_id":"",
                        "position_trace_id":"",
                        "position_confidence":0.0,
                        "position_session_info":{},
                        "position_session_metrics":{},
                        "position_interpreter":{},
                        "position_attribution":{},
                        "position_agent_direction":"",
                        "position_entry_owner_instance_id":"",
                        "position_entry_owner_strategy_id":"",
                        "position_current_manager_instance_id":"",
                        "position_current_manager_strategy_id":"",
                        "position_expected_exit_owner_instance_id":"",
                        "position_expected_exit_owner_strategy_id":"",
                        "position_exit_owner_instance_id":"",
                        "position_exit_owner_strategy_id":"",
                        "position_contributors":[],
                        "position_side":None,
                        "position_time":0,
                        "l1_breakeven_done": False,
                        "partial_done": False
                    })
                    _seal_close_guard(
                        state,
                        closed_side=str(state_before_close.get("position_side", "") or ""),
                        closed_ticket=int(state_before_close.get("position_ticket", 0) or 0),
                        close_action=action_upper,
                    )
                    _apply_realized_closure_stats(state, pnl, position_open=False)
                    # 同步更新旧的全局变量以防遗漏
                    global _last_close_time
                    _last_close_time = state["last_close_time"]
                    # 立即持久化 last_close_time，防止被后续 FILLED OPEN 覆盖或 save_state 遗漏
                    save_state(state)
                    # Phase 2: Record tactical feedback for EA FILLED close
                    _record_trade_feedback(state_before_close, pnl, fill_price, duration_min, action_upper, decision_id)
            elif status == "REJECTED":
                retcode = str(report.get("retcode", ""))
                action = report.get("action", "")
                modify_telemetry = state.get("_exit_modify_telemetry", {}) if isinstance(state.get("_exit_modify_telemetry", {}), dict) else {}
                if action == "MODIFY_SL":
                    modify_telemetry.pop("pending_sl_request_ts", None)
                    modify_telemetry.pop("pending_sl_value", None)
                    modify_telemetry.pop("pending_sl_signal_id", None)
                    modify_telemetry.pop("pending_sl_reason", None)
                    modify_telemetry.pop("pending_tp_after_sl_confirmed", None)
                    modify_telemetry.pop("pending_tp_after_sl_reason", None)
                    modify_telemetry.pop("pending_tp_after_sl_ticket", None)
                    state["_exit_modify_telemetry"] = modify_telemetry
                elif action == "MODIFY_TP":
                    modify_telemetry.pop("pending_tp_request_ts", None)
                    modify_telemetry.pop("pending_tp_value", None)
                    modify_telemetry.pop("pending_tp_signal_id", None)
                    modify_telemetry.pop("pending_tp_reason", None)
                    state["_exit_modify_telemetry"] = modify_telemetry
                _log(f"EA REJECTED: {retcode} {report.get('comment','')}", "ERROR")
                if resolved_decision_id:
                    _close_decision_state(
                        resolved_decision_id,
                        "FAILED",
                        reason=retcode or "EA_REJECTED",
                        signal_id=sig_id,
                        action=action,
                        retcode=retcode,
                        comment=report.get("comment", ""),
                    )
                if action == "CLOSE_ALL" and "UNKNOWN_ACTION" in retcode:
                    _log("[WARN]️ CLOSE_ALL 不受支持，自动回退为 CLOSE", "WARN")
                    write_trade_request(
                        "CLOSE",
                        state.get("position_side", "BUY"),
                        0, 0, 0,
                        vol=state.get("position_vol", 0.01),
                        sig_id=None,
                        request_origin="legacy_reject_fallback",
                    )
                    continue
                if action == "CLOSE_TICKET" and state.get("position_open"):
                    _log("[WARN]️ CLOSE_TICKET 被拒，自动回退为 CLOSE", "WARN")
                    write_trade_request(
                        "CLOSE",
                        state.get("position_side", "BUY"),
                        0, 0, 0,
                        vol=state.get("position_vol", 0.01),
                        sig_id=None,
                        request_origin="legacy_reject_fallback",
                    )
                    continue
                # PARTIAL_CLOSE_NOT_SUPPORTED fallback：券商不支持部分平仓时，全平以锁利
                if action == "CLOSE_PARTIAL":
                    state["partial_done"] = True
                    # PARTIAL_CLOSE_NOT_SUPPORTED：券商不支持部分平仓 → 全平锁利
                    if "PARTIAL_CLOSE_NOT_SUPPORTED" in retcode:
                        _log(f"[WARN]️ PARTIAL_CLOSE 不支持，执行全平 fallback", "WARN")
                    else:
                        _log(f"[WARN]️ PARTIAL_CLOSE 被拒 ({retcode})，标记已完成并执行全平 fallback", "WARN")
                    if state.get("position_open") and state.get("position_ticket"):
                        write_trade_request(
                            "CLOSE_TICKET",
                            state.get("position_side", "BUY"),
                            0, 0, 0,
                            vol=state.get("position_vol", 0.01),
                            sig_id=None,
                            ticket=state.get("position_ticket", 0),
                            request_origin="legacy_reject_fallback",
                        )
    except Exception as e: _daemon_log(f"process_ea_reports error: {e}")
    return state

# ============ 软挂单队列 (Soft Limits) ============
_soft_limits = {}  # 结构: {decision_id: {"side": "BUY", "limit_price": 1234.5, "sl": 1230, "tp": 1240, "vol": 0.05, "expire": time.time() + 1800, ...}}


def _remove_soft_limit(decision_id, to_state="VETOED", reason="SOFT_LIMIT_CLEARED", **metadata):
    global _soft_limits
    order = _soft_limits.pop(decision_id, None)
    if order is None:
        return None
    if to_state:
        _close_decision_state(
            decision_id,
            to_state,
            reason=reason,
            side=order.get("side"),
            limit_price=order.get("limit_price"),
            volume=order.get("vol"),
            entry_style=order.get("entry_style", "LIMIT"),
            **metadata,
        )
    return order


def _retire_legacy_soft_limits(state: Optional[dict] = None, *, reason: str = "PIPELINE_SINGLE_TRACK_SOFT_LIMIT_DISABLED") -> int:
    global _soft_limits
    if not _soft_limits:
        return 0
    retired = 0
    for sig_id in list(_soft_limits.keys()):
        removed = _remove_soft_limit(sig_id, to_state="VETOED", reason=reason)
        if removed is not None:
            retired += 1
    if retired > 0:
        current_state = state if isinstance(state, dict) else load_state()
        current_state["soft_limits"] = _soft_limits
        save_state(current_state)
        _daemon_log(f"[PIPELINE] retired {retired} legacy soft limits: reason={reason}")
    return retired


def _block_legacy_review_open(sig: dict, *, sig_id: str, stable_id: Optional[str] = None) -> bool:
    action = str(sig.get("action", "HOLD") or "HOLD").upper()
    if action not in {"OPEN", "BUY", "SELL"}:
        return False
    decision_id = str(sig.get("decision_id", stable_id or sig_id) or stable_id or sig_id)
    _close_decision_state(
        decision_id,
        "VETOED",
        reason="LEGACY_REVIEW_OPEN_DISABLED",
        review_signal_id=sig.get("signal_id", sig_id),
        stable_id=sig.get("stable_id", stable_id or sig_id),
        review_reason=sig.get("reason", ""),
    )
    try:
        state = load_state()
        _set_final_reject_reason(state, "LEGACY_REVIEW_OPEN_DISABLED", stage="review_approved_legacy_open")
        save_state(state)
    except Exception:
        pass
    _log("[WARN]️ Review 批准的旧开仓路径已停用，需改由 pipeline 统一执行", "WARN")
    return True

def _reanchor_sl_tp_by_design(side: str, exec_entry: float, design_entry: float, design_sl: float, design_tp: float):
    """
    用"原始设计距离"在最终执行价上重锚 SL/TP。
    目的：不论市价还是挂单触发，成交后都沿用同一套设计风险半径。
    """
    try:
        side_u = str(side or "").upper()
        entry_v = float(exec_entry or 0.0)
        design_entry_v = float(design_entry or 0.0)
        design_sl_v = float(design_sl or 0.0)
        design_tp_v = float(design_tp or 0.0)
    except Exception:
        return design_sl, design_tp

    if entry_v <= 0 or design_entry_v <= 0:
        return design_sl, design_tp

    sl_dist = abs(design_entry_v - design_sl_v) if design_sl_v > 0 else 0.0
    tp_dist = abs(design_tp_v - design_entry_v) if design_tp_v > 0 else 0.0
    if sl_dist <= 0 or tp_dist <= 0:
        return design_sl, design_tp

    if side_u == "BUY":
        return round(entry_v - sl_dist, 3), round(entry_v + tp_dist, 3)
    if side_u == "SELL":
        return round(entry_v + sl_dist, 3), round(entry_v - tp_dist, 3)
    return design_sl, design_tp


def _repair_sl_tp_for_rr(side: str, entry: float, sl: float, tp: float, atr: float = 0.0, min_rr: float = 1.0):
    """
    自动修复 SL/TP：
    1) 修复 BUY/SELL 的物理方向；
    2) 保障 TP/SL >= min_rr；
    3) 优先保留原始风险半径与目标，不做过度改写。
    """
    try:
        side_u = str(side or "").upper()
        entry_v = float(entry or 0.0)
        sl_v = float(sl or 0.0)
        tp_v = float(tp or 0.0)
        atr_v = max(0.0, float(atr or 0.0))
        rr_floor = max(1.0, float(min_rr or 1.0))
    except Exception:
        return sl, tp, False, 0.0

    if entry_v <= 0 or side_u not in {"BUY", "SELL"}:
        return sl_v, tp_v, False, 0.0

    min_stop = max(0.6, atr_v * 0.35)
    changed = False

    if side_u == "BUY":
        # 风险半径优先沿用原值，不合法时用 ATR/最小半径兜底
        sl_dist = entry_v - sl_v if 0 < sl_v < entry_v else abs(entry_v - sl_v)
        if sl_dist <= 0:
            sl_dist = max(min_stop, atr_v * 0.9)
        if sl_dist < min_stop:
            sl_dist = min_stop
        sl_new = entry_v - sl_dist
        tp_min = entry_v + sl_dist * rr_floor
        tp_candidate = tp_v if tp_v > entry_v else 0.0
        tp_new = max(tp_candidate, tp_min)
        changed = abs(sl_new - sl_v) > 1e-6 or abs(tp_new - tp_v) > 1e-6
    else:
        sl_dist = sl_v - entry_v if sl_v > entry_v > 0 else abs(sl_v - entry_v)
        if sl_dist <= 0:
            sl_dist = max(min_stop, atr_v * 0.9)
        if sl_dist < min_stop:
            sl_dist = min_stop
        sl_new = entry_v + sl_dist
        tp_min = entry_v - sl_dist * rr_floor
        tp_candidate = tp_v if 0 < tp_v < entry_v else 0.0
        tp_new = min(tp_candidate, tp_min) if tp_candidate > 0 else tp_min
        changed = abs(sl_new - sl_v) > 1e-6 or abs(tp_new - tp_v) > 1e-6

    rr_new = abs(tp_new - entry_v) / max(abs(sl_new - entry_v), 1e-6)
    return round(sl_new, 3), round(tp_new, 3), changed, rr_new


def _write_json_atomic(path: pathlib.Path, payload: dict) -> pathlib.Path:
    path.parent.mkdir(parents=True, exist_ok=True)
    last_err: Optional[Exception] = None
    for attempt in range(4):
        tmp_path = path.with_name(f".{path.name}.{os.getpid()}.{time.time_ns()}.tmp")
        try:
            with open(tmp_path, "w", encoding="utf-8") as handle:
                handle.write(json.dumps(payload, ensure_ascii=False, indent=2))
                handle.flush()
                os.fsync(handle.fileno())
            os.replace(tmp_path, path)
            return path
        except (PermissionError, OSError) as exc:
            last_err = exc
            try:
                if tmp_path.exists():
                    tmp_path.unlink()
            except Exception:
                pass
            time.sleep(0.15 * (2 ** attempt))
    # 4 次重试全部失败；最后再做一次 best-effort 写入并记录
    try:
        _daemon_log(f"[WRITE_JSON_ATOMIC] failed after retries: {path} err={last_err}")
    except Exception:
        pass
    return path


def _build_trade_request_intent_id(req: dict, *, existing_intent_id: str = "") -> str:
    if str(existing_intent_id or "").strip():
        return str(existing_intent_id).strip()
    try:
        from src.mt5_execution import generate_intent_id
    except ModuleNotFoundError:
        from mt5_execution import generate_intent_id
    signal_payload = {
        "signal_id": str(req.get("signal_id", "") or ""),
        "action": str(req.get("action", "HOLD") or "HOLD"),
        "side": str(req.get("side", "NONE") or "NONE"),
        "entry_price": float(req.get("entry_price", 0.0) or 0.0),
    }
    return str(generate_intent_id(signal_payload, ts_ms=int(time.time() * 1000)))


def _queue_post_fill_sl_tp_repair(
    *,
    ticket: int,
    side: str,
    fill_price: float,
    sl: float,
    tp: float,
    volume: float,
    decision_id: str = "",
    signal_id: str = "",
    attribution: Optional[dict] = None,
) -> bool:
    """成交后按真实 fill 价格复核 RR，必要时下发 SL/TP 修正请求。"""
    try:
        ticket_i = int(ticket or 0)
        fill_v = float(fill_price or 0.0)
        sl_v = float(sl or 0.0)
        tp_v = float(tp or 0.0)
        side_u = str(side or "").upper()
    except Exception:
        return False

    if ticket_i <= 0 or fill_v <= 0 or side_u not in {"BUY", "SELL"}:
        return False

    repaired_sl, repaired_tp, changed, repaired_rr = _repair_sl_tp_for_rr(
        side_u,
        fill_v,
        sl_v,
        tp_v,
        atr=0.0,
        min_rr=1.0,
    )
    if not changed:
        return False

    queued = False
    if abs(repaired_sl - sl_v) > 1e-6:
        sl_sig = f"postfill_{ticket_i}_MODIFY_SL"
        queued = bool(
            write_trade_request(
                "MODIFY_SL",
                side_u,
                fill_v,
                repaired_sl,
                repaired_tp,
                vol=volume,
                sig_id=sl_sig,
                ticket=ticket_i,
                decision_id=decision_id or signal_id or sl_sig,
                attribution=attribution if isinstance(attribution, dict) else None,
                request_origin="legacy_post_fill_repair",
            )
        ) or queued
    if abs(repaired_tp - tp_v) > 1e-6:
        tp_sig = f"postfill_{ticket_i}_MODIFY_TP"
        queued = bool(
            write_trade_request(
                "MODIFY_TP",
                side_u,
                fill_v,
                repaired_sl,
                repaired_tp,
                vol=volume,
                sig_id=tp_sig,
                ticket=ticket_i,
                decision_id=decision_id or signal_id or tp_sig,
                attribution=attribution if isinstance(attribution, dict) else None,
                request_origin="legacy_post_fill_repair",
            )
        ) or queued

    if queued:
        _daemon_log(
            f"[POST_FILL_RR_REPAIR] ticket={ticket_i} side={side_u} fill={fill_v:.2f} "
            f"SL {sl_v:.2f}->{repaired_sl:.2f} TP {tp_v:.2f}->{repaired_tp:.2f} RR={repaired_rr:.2f}"
        )
    return queued


def process_soft_limits(price_bid, price_ask):
    """Legacy soft limits are retired; pipeline owns pending-entry promotion."""
    if _pipeline_single_track_enabled():
        _retire_legacy_soft_limits(reason="PIPELINE_SINGLE_TRACK_SOFT_LIMIT_DISABLED")
        return False
    global _soft_limits
    now = time.time()
    triggered = []
    expired = []
    state_now = load_state()
    agent_dir_now, _agent_meta = get_agent_direction()

    for sig_id, order in list(_soft_limits.items()):
        if now > order["expire"]:
            expired.append(sig_id)
            continue
        registered_agent_dir = _soft_limit_registered_agent_direction(order)
        order_side = str(order.get("side", "") or "").upper()
        order_regime = str(order.get("regime", "UNKNOWN") or "UNKNOWN")
        if registered_agent_dir in ("LONG", "SHORT") and order_side in ("BUY", "SELL"):
            if _is_direction_conflict(registered_agent_dir, order_side):
                _daemon_log(
                    f"[BLOCK] [RegisteredDirection] 软挂单登记方向与订单方向冲突: "
                    f"registered={registered_agent_dir} side={order_side} limit={order.get('limit_price')}"
                )
                _remove_soft_limit(
                    sig_id,
                    to_state="REJECTED",
                    reason="SOFT_LIMIT_REGISTERED_DIRECTION_CONFLICT",
                    registered_agent_dir=registered_agent_dir,
                    signal_side=order_side,
                    limit_price=order.get("limit_price"),
                )
                continue
            if agent_dir_now in ("LONG", "SHORT") and agent_dir_now != registered_agent_dir:
                _daemon_log(
                    f"[BLOCK] [DirectionDrift] 软挂单方向漂移作废: registered={registered_agent_dir} "
                    f"current={str(agent_dir_now).upper()} side={order_side} limit={order.get('limit_price')}"
                )
                _remove_soft_limit(
                    sig_id,
                    to_state="REJECTED",
                    reason="SOFT_LIMIT_AGENT_DIRECTION_DRIFT",
                    registered_agent_dir=registered_agent_dir,
                    current_agent_dir=str(agent_dir_now).upper(),
                    signal_side=order_side,
                    limit_price=order.get("limit_price"),
                )
                continue
            
        if order["side"] == "BUY" and price_ask <= order["limit_price"]:
            cooldown_remaining = _cooldown_remaining_sec(
                state_now, regime=str(order.get("regime", "UNKNOWN") or "UNKNOWN")
            )
            if cooldown_remaining > 0:
                _daemon_log(
                    f"[BLOCK] [Cooldown] 软挂单触发被拦截(BUY): remaining={cooldown_remaining}s, limit={order['limit_price']}, ask={price_ask}"
                )
                continue
            if _is_direction_conflict(agent_dir_now, "BUY"):
                _daemon_log(
                    f"[BLOCK] [Direction] 软挂单触发被拦截(BUY): agent={str(agent_dir_now).upper()} limit={order['limit_price']}"
                )
                _remove_soft_limit(
                    sig_id,
                    to_state="REJECTED",
                    reason="DIRECTION_CONFLICT_HARD_BLOCK",
                    agent_dir=str(agent_dir_now).upper(),
                    signal_side="BUY",
                    trigger_price=price_ask,
                )
                continue
            triggered.append(sig_id)
            decision_id = str(order.get("decision_id", "") or sig_id)
            _daemon_log(f"[HIT] 软挂单(Buy Limit)触发: {price_ask} <= {order['limit_price']}")
            design_entry = float(order.get("design_entry", order.get("entry_ref", order["limit_price"])) or order["limit_price"])
            new_sl, new_tp = _reanchor_sl_tp_by_design("BUY", price_ask, design_entry, order.get("sl", 0), order.get("tp", 0))
            _transition_decision_state(
                decision_id,
                "EXECUTING",
                reason="SOFT_LIMIT_TRIGGERED",
                trigger_price=price_ask,
                limit_price=order["limit_price"],
                entry_style=order.get("entry_style", "LIMIT"),
                side=order["side"],
                design_entry=design_entry,
                anchored_sl=new_sl,
                anchored_tp=new_tp,
            )
            write_trade_request(
                "OPEN",
                "BUY",
                price_ask,
                new_sl,
                new_tp,
                order["vol"],
                sig_id,
                decision_id=decision_id,
                attribution=order.get("attribution"),
                request_origin="soft_limit_trigger",
            )
        elif order["side"] == "SELL" and price_bid >= order["limit_price"]:
            cooldown_remaining = _cooldown_remaining_sec(
                state_now, regime=str(order.get("regime", "UNKNOWN") or "UNKNOWN")
            )
            if cooldown_remaining > 0:
                _daemon_log(
                    f"[BLOCK] [Cooldown] 软挂单触发被拦截(SELL): remaining={cooldown_remaining}s, limit={order['limit_price']}, bid={price_bid}"
                )
                continue
            if _is_direction_conflict(agent_dir_now, "SELL"):
                _daemon_log(
                    f"[BLOCK] [Direction] 软挂单触发被拦截(SELL): agent={str(agent_dir_now).upper()} limit={order['limit_price']}"
                )
                _remove_soft_limit(
                    sig_id,
                    to_state="REJECTED",
                    reason="DIRECTION_CONFLICT_HARD_BLOCK",
                    agent_dir=str(agent_dir_now).upper(),
                    signal_side="SELL",
                    trigger_price=price_bid,
                )
                continue
            triggered.append(sig_id)
            decision_id = str(order.get("decision_id", "") or sig_id)
            _daemon_log(f"[HIT] 软挂单(Sell Limit)触发: {price_bid} >= {order['limit_price']}")
            design_entry = float(order.get("design_entry", order.get("entry_ref", order["limit_price"])) or order["limit_price"])
            new_sl, new_tp = _reanchor_sl_tp_by_design("SELL", price_bid, design_entry, order.get("sl", 0), order.get("tp", 0))
            _transition_decision_state(
                decision_id,
                "EXECUTING",
                reason="SOFT_LIMIT_TRIGGERED",
                trigger_price=price_bid,
                limit_price=order["limit_price"],
                entry_style=order.get("entry_style", "LIMIT"),
                side=order["side"],
                design_entry=design_entry,
                anchored_sl=new_sl,
                anchored_tp=new_tp,
            )
            write_trade_request(
                "OPEN",
                "SELL",
                price_bid,
                new_sl,
                new_tp,
                order["vol"],
                sig_id,
                decision_id=decision_id,
                attribution=order.get("attribution"),
                request_origin="soft_limit_trigger",
            )
            
    for sig_id in expired:
        _daemon_log(f"⏳ 软挂单过期清理: {sig_id}")
        _remove_soft_limit(sig_id, to_state="VETOED", reason="SOFT_LIMIT_EXPIRED", expired_at=now)
    for sig_id in triggered:
        _soft_limits.pop(sig_id, None)
        
    # 同步状态，让前端立刻知道挂单被清理了
    if len(triggered) > 0 or len(expired) > 0:
        # 这里只需要简单返回一个信号让主循环去 save_state 即可
        pass
    return len(triggered) > 0

_last_rejected_signal = None
_last_rejected_time = 0

def write_trade_request(
    action,
    side,
    entry,
    sl,
    tp,
    vol=0.01,
    sig_id=None,
    ticket=0,
    decision_id=None,
    stable_id=None,
    attribution=None,
    request_origin: str = "bridge_legacy",
    legacy_fallback: bool = False,
    intent_id: str = "",
):
    global _last_rejected_signal, _last_rejected_time
    
    sig_id = sig_id or f"sig_{time.strftime('%Y%m%d_%H%M%S')}"
    decision_id = str(decision_id or sig_id)
    action_upper = str(action or "").upper()
    request_origin = str(request_origin or "bridge_legacy")
    if _block_legacy_trade_request(
        action_upper,
        request_origin,
        decision_id=decision_id,
        side=str(side or ""),
    ):
        return None
    if (
        _pipeline_single_track_enabled()
        and action_upper == "OPEN"
        and request_origin not in {"pipeline_dispatch", "pipeline_pending_promotion"}
    ):
        _daemon_log(
            f"[PIPELINE_OWNER] blocked non-pipeline OPEN request origin={request_origin} sig={sig_id}"
        )
        return None
    # ── HOLD 短路：EA 不支持 HOLD，禁止写入 to_ea ──
    if action_upper not in {"OPEN", "CLOSE", "CLOSE_TICKET", "MODIFY_SL", "MODIFY_TP", "CLOSE_PARTIAL", "REDUCE"}:
        _daemon_log(f"[AUDIT] write_trade_request skipped: unsupported action={action_upper} sig={sig_id}")
        return None
    if action_upper == "OPEN":
        # ── 多仓位上限拦截（替代旧版单一持仓硬拦截）──
        state_preopen = load_state()
        positions_all = state_preopen.get("positions_all", [])
        if not isinstance(positions_all, list):
            positions_all = []
        current_positions = len(positions_all)
        # 回退兼容：旧版 state 没有 positions_all 但有 position_open 标记
        if current_positions == 0 and bool(state_preopen.get("position_open", False)):
            current_positions = 1
        max_positions = min(5, int(
            _get_effective_param("max_open_positions",
            _get_effective_param("max_positions", 4)) or 4))
        if current_positions >= max_positions:
            _daemon_log(
                f"[AUDIT] max positions reached "
                f"(current={current_positions} max={max_positions}), "
                f"blocking OPEN {side}"
            )
            return None

        try:
            entry = float(entry or 0)
            sl = float(sl or 0)
            tp = float(tp or 0)
            # 兜底修复：任何 OPEN 在落盘前都先修复一次 RR/方向，避免错误参数穿透到 EA。
            sl, tp, repaired, repaired_rr = _repair_sl_tp_for_rr(side, entry, sl, tp, atr=0.0, min_rr=1.5)
            if repaired:
                _daemon_log(
                    f"[OPEN_REPAIR] side={side} entry={entry:.2f} repaired SL/TP -> "
                    f"SL={sl:.2f} TP={tp:.2f} RR={repaired_rr:.2f}"
                )
            
            # ── 熔断去重机制：防止相同倒挂信号短时间内疯狂刷屏 ──
            current_sig_hash = f"{side}_{entry}_{sl}_{tp}"
            now = time.time()
            
            # ── 硬性拦截：SL=0 或 TP=0 的开仓单永不写入，直接抛异常 ──
            if sl <= 0 or tp <= 0:
                if current_sig_hash != _last_rejected_signal or (now - _last_rejected_time) > 300:
                    msg = f"[AUDIT [FAIL]] 致命拦截：尝试写入 SL={sl} 或 TP={tp} 的裸奔订单，已强制阻断！"
                    _log(msg, "ERROR")
                    _daemon_log(msg)
                    _last_rejected_signal = current_sig_hash
                    _last_rejected_time = now
                return None
                    
            # ── 新的硬性拦截：禁止发向 MT5 的任何倒挂订单 ──
            if side == "BUY" and (sl >= entry or tp <= entry):
                if current_sig_hash != _last_rejected_signal or (now - _last_rejected_time) > 300:
                    msg = f"[AUDIT [FAIL]] 致命拦截：BUY 单倒挂 (entry={entry}, sl={sl}, tp={tp})，已强制阻断！"
                    _log(msg, "ERROR")
                    _daemon_log(msg)
                    _last_rejected_signal = current_sig_hash
                    _last_rejected_time = now
                return None
            if side == "SELL" and (sl <= entry or tp >= entry):
                if current_sig_hash != _last_rejected_signal or (now - _last_rejected_time) > 300:
                    msg = f"[AUDIT [FAIL]] 致命拦截：SELL 单倒挂 (entry={entry}, sl={sl}, tp={tp})，已强制阻断！"
                    _log(msg, "ERROR")
                    _daemon_log(msg)
                    _last_rejected_signal = current_sig_hash
                    _last_rejected_time = now
                return None

            # ── 冷却硬闸门：任何 OPEN 请求都必须通过，防止软挂单等旁路绕过 ──
            state_cooldown = load_state()
            strategy_id_for_cooldown = ""
            regime_for_cooldown = "UNKNOWN"
            if isinstance(attribution, dict):
                strategy_id_for_cooldown = str(attribution.get("strategy_id", "") or "")
                regime_for_cooldown = str(attribution.get("regime", state_cooldown.get("position_regime", "UNKNOWN")) or "UNKNOWN")
            cooldown_remaining = _cooldown_remaining_sec(
                state_cooldown, strategy_id=strategy_id_for_cooldown, regime=regime_for_cooldown
            )
            if cooldown_remaining > 0:
                if current_sig_hash != _last_rejected_signal or (now - _last_rejected_time) > 30:
                    msg = (
                        f"[AUDIT [BLOCK]] 冷却拦截：OPEN {side} 被阻断，remaining={cooldown_remaining}s "
                        f"(entry={entry}, vol={float(vol or 0):.2f})"
                    )
                    _log(msg, "WARN")
                    _daemon_log(msg)
                    _last_rejected_signal = current_sig_hash
                    _last_rejected_time = now
                return None

            strat_id = str((attribution or {}).get("strategy_id", "") or "")
            reentry_remaining, reentry_reason = _same_side_reentry_veto(state_cooldown, side, strategy_id=strat_id)
            if reentry_remaining > 0:
                if current_sig_hash != _last_rejected_signal or (now - _last_rejected_time) > 30:
                    msg = (
                        f"[AUDIT [BLOCK]] 同向瞬时重开拦截：OPEN {side} 被阻断，remaining={reentry_remaining}s "
                        f"reason={reentry_reason} (entry={entry}, vol={float(vol or 0):.2f})"
                    )
                    _log(msg, "WARN")
                    _daemon_log(msg)
                    _last_rejected_signal = current_sig_hash
                    _last_rejected_time = now
                return None

            regime_for_gate = str((attribution or {}).get("regime", state_cooldown.get("position_regime", "UNKNOWN")) or "UNKNOWN")
            current_agent_dir, _ = get_agent_direction()
            current_agent_dir = str(current_agent_dir or "").upper()
            registered_agent_dir = str((attribution or {}).get("agent_direction", "") or "").upper()
            effective_agent_dir = current_agent_dir if current_agent_dir in ("LONG", "SHORT") else registered_agent_dir
            if effective_agent_dir in ("LONG", "SHORT") and side in ("BUY", "SELL"):
                if _is_direction_conflict(effective_agent_dir, side):
                    if current_sig_hash != _last_rejected_signal or (now - _last_rejected_time) > 30:
                        msg = (
                            f"[AUDIT [BLOCK]] 方向终审拦截：OPEN {side} 被阻断，agent={effective_agent_dir} "
                            f"(current={current_agent_dir or 'N/A'}, registered={registered_agent_dir or 'N/A'}, regime={regime_for_gate})"
                        )
                        _log(msg, "ERROR")
                        _daemon_log(msg)
                        _last_rejected_signal = current_sig_hash
                        _last_rejected_time = now
                    if decision_id:
                        _transition_decision_state(
                            decision_id,
                            "REJECTED",
                            reason="DIRECTION_CONFLICT_HARD_BLOCK",
                            agent_dir=effective_agent_dir,
                            signal_side=side,
                            registered_agent_dir=registered_agent_dir,
                            current_agent_dir=current_agent_dir,
                            regime=regime_for_gate,
                        )
                    return None
                
        except Exception as e:
            _daemon_log(f"write_trade_request audit failed: {e}")

        # ── Multi-strategy integration: arbitrator + allocator + sentinel ──
        try:
            from src.signal_arbitrator import arbitrate
            from src.capital_allocator import allocate
            from src.sentinel_deployer import should_deploy, deploy_sentinel

            # Arbitrator: check if this signal conflicts with existing active signals
            if attribution and isinstance(attribution, dict):
                strategy_id = str(attribution.get("strategy_id", "") or "")
                if strategy_id:
                    regime = str((attribution or {}).get("regime", state_cooldown.get("position_regime", "UNKNOWN")) or "UNKNOWN")
                    arb_result = arbitrate(
                        [{"strategy_id": strategy_id, "side": side, "confidence": 0.7,
                          "timeframe": str(attribution.get("timeframe", "M5"))}],
                        regime=regime,
                        max_positions=int(_get_effective_param("max_positions", 1) or 1),
                    )
                    if len(arb_result.get("accepted", [])) == 0:
                        _daemon_log(f"[ARBITRATOR] signal from {strategy_id} rejected by regime gate ({regime})")
                        # Don't hard-block — let existing audit chain handle it

            # Capital allocator: compute position size
            alloc_result = allocate(
                [{"strategy_id": str(attribution.get("strategy_id", "unknown")),
                  "strategy_type": str(attribution.get("strategy_style", "trend")),
                  "base_weight": 0.02}],
                {"trend": 0.55, "mean_reversion": 0.30, "sentinel": 0.15},
            )
            alloc_vol = alloc_result.get("allocations", {}).get(
                str(attribution.get("strategy_id", "unknown")), 0.01) if isinstance(attribution, dict) else 0.01
            if alloc_vol > 0 and alloc_vol < float(vol or 0.01):
                vol = alloc_vol  # Use allocator's recommended size

        except Exception:
            pass
        # ── End multi-strategy integration ──

        try:
            request_strategy_id = canonical_strategy_id(
                (attribution or {}).get("strategy_id", "") if isinstance(attribution, dict) else ""
            )

            def _same_open_fingerprint(payload):
                payload_action = str(payload.get("action", "")).upper()
                payload_side = str(payload.get("side", "")).upper()
                payload_entry = float(payload.get("entry_price", 0) or 0)
                payload_sl = float(payload.get("sl_price", 0) or 0)
                payload_tp = float(payload.get("tp_price", 0) or 0)
                payload_vol = float(payload.get("volume", 0) or 0)
                payload_attribution = payload.get("attribution", {}) if isinstance(payload.get("attribution", {}), dict) else {}
                payload_strategy_id = canonical_strategy_id(
                    payload.get("strategy_id", "") or payload_attribution.get("strategy_id", "")
                )
                return (
                    payload_action == "OPEN"
                    and payload_side == str(side).upper()
                    and payload_strategy_id == request_strategy_id
                    and abs(payload_entry - entry) <= 0.35
                    and abs(payload_sl - sl) <= 0.35
                    and abs(payload_tp - tp) <= 0.35
                    and abs(payload_vol - float(vol or 0)) <= 1e-6
                )

            state_now = load_state()
            for p in state_now.get("positions_all", []) or []:
                p_side = str(p.get("side", "")).upper()
                p_entry = float(p.get("entry", 0) or 0)
                p_vol = float(p.get("volume", 0) or 0)
                p_strategy_id = canonical_strategy_id(p.get("strategy_id", ""))
                if request_strategy_id and p_side == str(side).upper() and p_strategy_id == request_strategy_id:
                    _daemon_log(
                        f"Skip duplicate OPEN: same strategy same side already open "
                        f"strategy={request_strategy_id} side={side}"
                    )
                    return None
                if p_side == str(side).upper() and abs(p_entry - entry) <= 0.35 and abs(p_vol - float(vol or 0)) <= 1e-6:
                    _daemon_log(f"Skip duplicate OPEN: existing MT5 position side={side} entry={p_entry} vol={p_vol}")
                    return None

            if _bridge_to_ea_dir().exists():
                for req_file in sorted(_bridge_to_ea_dir().glob("*.request.json"), key=lambda x: x.stat().st_mtime, reverse=True)[:20]:
                    try:
                        if time.time() - req_file.stat().st_mtime > 180:
                            continue
                        existing_req = json.loads(req_file.read_text(encoding="utf-8"))
                        if _same_open_fingerprint(existing_req):
                            _daemon_log(f"Skip duplicate OPEN: pending request {req_file.name}")
                            return None
                    except Exception:
                        pass

            if ARCHIVE_DIR.exists():
                for report_file in sorted(ARCHIVE_DIR.glob("*.report*.json"), key=lambda x: x.stat().st_mtime, reverse=True)[:50]:
                    try:
                        if time.time() - report_file.stat().st_mtime > 180:
                            continue
                        report = json.loads(report_file.read_text(encoding="utf-8"))
                        payload = {
                            "action": report.get("action"),
                            "side": report.get("side"),
                            "strategy_id": (report.get("attribution", {}) or {}).get("strategy_id", ""),
                            "entry_price": entry,
                            "sl_price": report.get("sl_price", 0),
                            "tp_price": report.get("tp_price", 0),
                            "volume": report.get("volume", 0),
                        }
                        if _same_open_fingerprint(payload):
                            _daemon_log(f"Skip duplicate OPEN: recent report {report_file.name}")
                            return None
                    except Exception:
                        pass
        except Exception as e:
            _daemon_log(f"duplicate OPEN guard failed: {e}")
    req = {"signal_id": sig_id, "decision_id": decision_id, "action": action_upper, "side": side, "symbol": _detect_active_symbol(),
           "volume": vol, "entry_price": entry, "sl_price": sl, "tp_price": tp,
           "reason": "auto", "confidence": 0.80, "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ")}
    try:
        lot_step = float(_get_effective_param("lot_step", 0.01) or 0.01)
        min_lot = float(_get_effective_param("min_lot", lot_step) or lot_step)
        v = float(req.get("volume", vol) or vol or 0.0)
        if lot_step > 0 and v > 0:
            v = round(round(v / lot_step) * lot_step, 2)
        if v > 0:
            v = max(min_lot, v)
        req["volume"] = round(v, 2)
    except Exception:
        pass
    if stable_id:
        req["stable_id"] = stable_id
    if ticket:
        req["ticket"] = int(ticket)
    req["request_origin"] = str(request_origin or "bridge_legacy")
    req["legacy_fallback"] = bool(legacy_fallback)
    req["intent_id"] = _build_trade_request_intent_id(req, existing_intent_id=intent_id)
    try:
        _bridge_to_ea_dir().mkdir(parents=True, exist_ok=True)
        state_now = load_state()
        if not isinstance(attribution, dict) or not attribution:
            if action in {"CLOSE", "REDUCE"}:
                attribution = _build_trade_attribution(state=state_now)
            else:
                attribution = {}
        normalized_attribution = (
            normalize_position_attribution(
                attribution,
                action=action,
                fallback=state_now.get("position_attribution", {}) if isinstance(state_now.get("position_attribution", {}), dict) else {},
            )
            if isinstance(attribution, dict) and attribution
            else {}
        )
        if normalized_attribution:
            req["attribution"] = normalized_attribution
        if action_upper == "OPEN" and normalized_attribution:
            state_now.update(position_state_fields_from_attribution(normalized_attribution, action="OPEN"))
            save_state(state_now)
            sync_open_request(normalized_attribution, data_dir=DATA_DIR, reason="bridge_write_trade_request")
        elif action_upper in {"CLOSE", "REDUCE"} and normalized_attribution:
            state_now.update(position_state_fields_from_attribution(normalized_attribution, action="CLOSE"))
            save_state(state_now)
            sync_close_request(
                normalized_attribution,
                data_dir=DATA_DIR,
                ticket=int(ticket or state_now.get("position_ticket", 0) or 0),
                reason="bridge_write_trade_request",
            )
        # 使用原子重命名，防止 EA 读到半截文件
        target_path = _bridge_to_ea_dir() / f"{sig_id}.request.json"
        _write_json_atomic(target_path, req)
        _append_decision_link(
            signal_id=sig_id,
            decision_id=decision_id,
            action=action,
            stable_id=str(stable_id or ""),
            ticket=ticket,
            source="write_trade_request",
            request_origin=req.get("request_origin", ""),
            legacy_fallback=req.get("legacy_fallback", False),
            intent_id=req.get("intent_id", ""),
        )
        
        _daemon_log(f"Request: {action_upper} {side} @{entry}")
        return sig_id
    except Exception as e: _daemon_log(f"Write req failed: {e}"); return None


def trigger_panic_close_and_block(reason: str = "MANUAL_PANIC", source: str = "gui") -> dict:
    """人工紧急平仓: 立即清空软挂单、清理待执行 OPEN 请求，并请求当前持仓全平。"""
    global _soft_limits

    state = load_state()
    now_ts = time.time()

    cleared_soft_limits = 0
    for sig_id in list(_soft_limits.keys()):
        _remove_soft_limit(sig_id, to_state="VETOED", reason="KILL_SWITCH_PANIC")
        cleared_soft_limits += 1

    removed_open_requests = 0
    try:
        for req_file in list(_bridge_to_ea_dir().glob("*.request.json")):
            try:
                payload = json.loads(req_file.read_text(encoding="utf-8"))
            except Exception:
                continue
            if str(payload.get("action", "") or "").upper() != "OPEN":
                continue
            req_file.unlink(missing_ok=True)
            removed_open_requests += 1
    except Exception as e:
        _daemon_log(f"[PANIC] purge pending OPEN requests failed: {e}")

    state["soft_limits"] = _soft_limits
    save_state(state)

    mt5_state = get_mt5_position_state()
    _update_account_state(mt5_state.get("balance"), mt5_state.get("equity"))
    positions_all = mt5_state.get("positions_all", []) if isinstance(mt5_state.get("positions_all", []), list) else []
    detected_positions = positions_all or state.get("positions_all", []) or []
    # 从 detected_positions 中提取首个有效 side/volume
    _first_pos_side = "BUY"
    _first_pos_vol = 0.01
    if detected_positions:
        _fp = detected_positions[0]
        if isinstance(_fp, dict):
            _first_pos_side = str(_fp.get("side", "BUY") or "BUY").upper()
            try:
                _first_pos_vol = float(_fp.get("volume", 0.0) or 0.0)
            except Exception:
                _first_pos_vol = 0.01
            if _first_pos_vol <= 0:
                _first_pos_vol = 0.01

    active_ticket = int(
        mt5_state.get("position_ticket", 0)
        or state.get("position_ticket", 0)
        or 0
    )
    active_side = (
        mt5_state.get("position_side")
        or state.get("position_side")
        or _first_pos_side
    )
    active_vol = float(
        mt5_state.get("position_vol", 0.0)
        or state.get("position_vol", 0.0)
        or _first_pos_vol
    )

    close_requested = False
    requested_actions = []
    has_live_position = bool(mt5_state.get("position_open")) or bool(state.get("position_open", False)) or bool(detected_positions)
    if has_live_position:
        close_all_sig = f"panic_close_all_{int(now_ts)}"
        close_requested = (
            write_trade_request(
                "CLOSE",
                active_side or "BUY",
                0,
                0,
                0,
                vol=active_vol,
                sig_id=close_all_sig,
                decision_id=state.get("position_decision_id") or close_all_sig,
                request_origin="manual_panic",
            )
            is not None
        )
        if close_requested:
            requested_actions.append("CLOSE")
            _log(
                f"[PANIC] Panic close issued: action=CLOSE positions={len(detected_positions) or 1} side={active_side} vol={active_vol}",
                "ERROR",
            )
        if active_ticket > 0:
            close_ticket_sig = f"panic_close_ticket_{active_ticket}_{int(now_ts)}"
            close_ticket_requested = (
                write_trade_request(
                    "CLOSE_TICKET",
                    active_side or "BUY",
                    0,
                    0,
                    0,
                    vol=active_vol,
                    sig_id=close_ticket_sig,
                    decision_id=state.get("position_decision_id") or close_ticket_sig,
                    ticket=active_ticket,
                    request_origin="manual_panic",
                )
                is not None
            )
            close_requested = close_requested or close_ticket_requested
            if close_ticket_requested:
                requested_actions.append("CLOSE_TICKET")

    message = (
        f"紧急阻断已生效: kill_switch=ON, soft_limits_cleared={cleared_soft_limits}, "
        f"pending_open_requests_cleared={removed_open_requests}, close_requested={close_requested}, "
        f"requested_actions={','.join(requested_actions) or 'NONE'}, detected_positions={len(detected_positions)}"
    )
    _daemon_log(f"[PANIC] {message}")
    return {
        "success": True,
        "message": message,
        "kill_switch": True,
        "soft_limits_cleared": cleared_soft_limits,
        "pending_open_requests_cleared": removed_open_requests,
        "close_requested": close_requested,
        "requested_actions": requested_actions,
        "detected_positions": len(detected_positions),
        "position_ticket": active_ticket,
    }

# ============ 主循环 ============
def reconcile_positions(state):
    """
    状态对账与异常接管：对比 MT5 真实持仓和本地 SQLite 记录
    如果发现 MT5 有仓位但本地没有记录（孤儿单），则接管并写入本地状态。
    如果发现本地有记录但 MT5 没仓位（被手动平仓或止盈止损），则清理本地状态。
    同时清理 orphaned journal 文件。
    """
    mt5_state = get_mt5_position_state()
    if mt5_state.get("error"):
        # MT5 files are unreadable — advance ghost counter if we track a position
        # so that extended EA outages don't permanently mask ghost orders.
        db_ticket = int(state.get("position_ticket", 0) or 0)
        if db_ticket > 0:
            ghost_count = int(state.get("ghost_check_count", 0)) + 1
            state["ghost_check_count"] = ghost_count
            _log(
                f"[WARN]️ MT5 读取失败 ({mt5_state['error']})，对账跳过但幽灵计数器推进: "
                f"ticket={db_ticket} count={ghost_count}/3",
                "WARN",
            )
        else:
            _log(f"[WARN]️ 无法读取 MT5 持仓状态 ({mt5_state['error']})，本次对账跳过", "WARN")
        return state
    _update_account_state(mt5_state.get("balance"), mt5_state.get("equity"))

    mt5_positions = mt5_state.get("positions_all", [])
    mt5_tickets = {p["ticket"] for p in mt5_positions}

    persist_after_reconcile = False
    try:
        conn = _open_state_db(timeout=10.0)
        c = conn.cursor()

        # ── 1. 读取 state_kv 中的持仓状态（权威数据源）──
        c.execute("SELECT value FROM state_kv WHERE key = 'position_ticket'")
        row = c.fetchone()
        db_ticket = int(row[0]) if row and row[0] else 0

        # 修复点：确保我们使用的是全局的 mt5_positions 而不是局部空表
        # 因为 MT5_FILES 轮询机制下，get_mt5_position_state 返回的 positions_all 才是最新的。

        # ── 2. 票号替换：MT5 仍有仓，但当前票号已经替换掉本地旧票 ──
        if db_ticket and mt5_tickets and db_ticket not in mt5_tickets and len(mt5_tickets) == 1:
            rollover_ticket = next(iter(mt5_tickets))
            rollover_position = next((p for p in mt5_positions if p["ticket"] == rollover_ticket), None)
            if rollover_position:
                lifecycle_current = {}
                try:
                    lifecycle_payload = load_position_lifecycle(DATA_DIR)
                    current_payload = (
                        lifecycle_payload.get("current", {})
                        if isinstance(lifecycle_payload.get("current", {}), dict)
                        else {}
                    )
                    lifecycle_status = str(current_payload.get("status", "") or "").strip().lower()
                    lifecycle_ticket = int(current_payload.get("ticket", 0) or 0)
                    if lifecycle_status in {"open", "opening", "closing"} and lifecycle_ticket == db_ticket:
                        lifecycle_current = current_payload
                except Exception:
                    lifecycle_current = {}

                rollover_side = str(rollover_position.get("side", "BUY") or "BUY").upper()
                rollover_entry = float(rollover_position.get("entry", 0.0) or 0.0)
                rollover_sl = float(rollover_position.get("sl", 0.0) or 0.0)
                rollover_tp = float(rollover_position.get("tp", 0.0) or 0.0)
                rollover_volume = float(rollover_position.get("volume", 0.0) or 0.0)

                rollover_attribution = normalize_position_attribution(
                    state.get("position_attribution", {}) if isinstance(state.get("position_attribution", {}), dict) else {},
                    action="OPEN",
                    fallback=lifecycle_current or state,
                )
                state.update(
                    {
                        "position_open": True,
                        "position_side": rollover_side,
                        "position_ticket": rollover_ticket,
                        "position_entry": rollover_entry,
                        "position_price": rollover_entry,
                        "position_sl": rollover_sl,
                        "position_tp": rollover_tp,
                        "position_vol": rollover_volume,
                        "position_time": time.time(),
                        "positions_all": mt5_positions,
                        "ghost_check_count": 0,
                        "partial_done": False,
                        "orphan_recovered": True,
                    }
                )
                state.update(
                    position_state_fields_from_attribution(
                        rollover_attribution,
                        action="OPEN",
                        fallback=lifecycle_current or state,
                    )
                )
                try:
                    sync_open_fill(
                        rollover_attribution,
                        data_dir=DATA_DIR,
                        ticket=int(rollover_ticket or 0),
                        side=rollover_side,
                        volume=float(rollover_volume or 0.0),
                        entry_price=float(rollover_entry or 0.0),
                        sl=float(rollover_sl or 0.0),
                        tp=float(rollover_tp or 0.0),
                        symbol=str(rollover_position.get("symbol", "XAUUSD") or "XAUUSD"),
                        source="bridge_reconcile_ticket_rollover",
                    )
                except Exception as lifecycle_err:
                    _daemon_log(f"sync_open_fill ticket rollover failed: {lifecycle_err}")
                state.get("orphan_check_counts", {}).pop(f"orphan_{rollover_ticket}", None)
                persist_after_reconcile = True
                _log(
                    f"Reconciliation: Ticket rollover synchronized {db_ticket}->{rollover_ticket} from MT5.",
                    "WARN",
                )
                db_ticket = rollover_ticket

        # ── 3. 幽灵单：state_kv 说有持仓，但 MT5 实际没有（被外部平仓/止损）──
        # 加入防误杀机制：如果连续 N 次发现 MT5 没单，才真正清理。防止 EA 重启瞬间的数据为空导致误杀。
        if db_ticket and db_ticket not in mt5_tickets:
            ghost_count = int(state.get("ghost_check_count", 0)) + 1
            state["ghost_check_count"] = ghost_count
            
            if ghost_count >= 3: # 连续 3 次 (约 30 秒) 都没在 MT5 看到这笔单子，才判定为幽灵单
                _log(f"Reconciliation: Ghost ticket={db_ticket} in DB but not in MT5 (checked {ghost_count} times). Cleaning.", "WARN")
                state_before_close = dict(state)
                # 强制设置状态并在内存和数据库中落盘，防止被并发覆盖
                state["position_open"] = False
                save_state(state)
                
                try:
                    sync_flat_recovery(
                        data_dir=DATA_DIR,
                        reason="GHOST_RECONCILIATION_STATE_ONLY",
                        source="bridge_reconcile_ghost",
                    )
                except Exception as lifecycle_err:
                    _daemon_log(f"sync_flat_recovery ghost reconcile failed: {lifecycle_err}")
                c.execute("DELETE FROM state_kv WHERE key IN (?, ?, ?, ?, ?, ?, ?, ?)",
                          ("position_open", "position_side", "position_ticket",
                           "position_entry", "position_price", "position_sl", "position_tp", "position_vol"))
                # 清理幽灵单后，重置计数器
                state["ghost_check_count"] = 0
                conn.commit()
                # 只有这里修改 state 对象
                state.update({
                    "position_open": False,
                    "position_ticket": 0,
                    "position_time": 0,
                    "position_signal_id": "",
                    "position_decision_id": "",
                    "position_regime": "UNKNOWN",
                    "position_strategy_id": "",
                    "position_trace_id": "",
                    "position_confidence": 0.0,
                    "position_session_info": {},
                    "position_session_metrics": {},
                    "position_interpreter": {},
                    "position_attribution": {},
                    "position_agent_direction": "",
                    "position_entry_owner_instance_id": "",
                    "position_entry_owner_strategy_id": "",
                    "position_current_manager_instance_id": "",
                    "position_current_manager_strategy_id": "",
                    "position_expected_exit_owner_instance_id": "",
                    "position_expected_exit_owner_strategy_id": "",
                    "position_exit_owner_instance_id": "",
                    "position_exit_owner_strategy_id": "",
                    "position_contributors": [],
                    "recent_closures": state_before_close.get("recent_closures", []),
                    "pattern_memory": state_before_close.get("pattern_memory", {}),
                    "runtime_adaptation": state_before_close.get("runtime_adaptation", {}),
                    "learning_adaptation_summary": state_before_close.get("learning_adaptation_summary", {}),
                })
                persist_after_reconcile = True
                _log("Reconciliation: Ghost position cleared from state_kv without synthetic pnl/stat updates", "WARN")
            else:
                _log(f"Reconciliation: Potential ghost ticket={db_ticket} (check {ghost_count}/3). Waiting for MT5 sync.", "INFO")
                save_state(state)
        elif db_ticket:
            if int(state.get("ghost_check_count", 0)) > 0:
                state["ghost_check_count"] = 0
                save_state(state)

        # ── 4. 孤儿单：MT5 有，但 state_kv 没有（断电前开仓刚好在启动期间成交，或因 Bug 漏记）──
        # 对称于幽灵单的防误杀机制：连续 N 次确认 MT5 有该单子才接管，防止 EA 重启瞬间
        # 数据闪烁导致的错误接管。
        if mt5_tickets and not db_ticket:
            t = next(iter(mt5_tickets))
            p_info = next((p for p in mt5_positions if p["ticket"] == t), None)
            if p_info:
                orphan_key = f"orphan_{t}"
                orphan_check_count = int(state.get("orphan_check_counts", {}).get(orphan_key, 0)) + 1
                state.setdefault("orphan_check_counts", {})[orphan_key] = orphan_check_count

                if orphan_check_count < 3:
                    _log(f"Reconciliation: Potential orphan ticket={t} (check {orphan_check_count}/3). Waiting for MT5 sync.", "INFO")
                    save_state(state)
                else:
                    # 修复孤儿单接管逻辑，确保完全同步 MT5 的所有字段
                    side_str = p_info["side"]
                    entry_price = float(p_info["entry"])
                    current_sl = float(p_info.get("sl", 0))
                    current_tp = float(p_info.get("tp", 0))
                    current_vol = float(p_info.get("volume", 0.01))

                    keys = ["position_open","position_side","position_ticket","position_entry",
                            "position_price","position_sl","position_tp","position_vol","position_time"]
                    vals = ["true", f'"{side_str}"', str(t),
                            str(entry_price), str(entry_price),
                            str(current_sl), str(current_tp),
                            str(current_vol),
                            str(int(time.time()))]

                    c.execute("DELETE FROM state_kv WHERE key IN (?, ?, ?, ?, ?, ?, ?, ?, ?)", keys)
                    for k, v in zip(keys, vals):
                        c.execute("INSERT INTO state_kv (key, value, updated_at) VALUES (?, ?, datetime('now'))", [k, v])
                    conn.commit()

                    # 同步更新内存中的 state
                    # 注意：position_time 使用孤儿单恢复时间（time.time()），
                    # 且标记为孤儿单以区分真实开仓时间（用于时间止损判断）
                    state.update({
                        "position_open": True,
                        "position_side": side_str,
                        "position_ticket": t,
                        "position_entry": entry_price,
                        "position_sl": current_sl,
                        "position_tp": current_tp,
                        "position_vol": current_vol,
                        "position_time": time.time(),
                        "orphan_recovered": True,   # 孤儿单标记：时间止损应宽容处理
                        "partial_done": False
                    })
                    try:
                        orphan_attribution = normalize_position_attribution(
                            state.get("position_attribution", {}) if isinstance(state.get("position_attribution", {}), dict) else {},
                            action="OPEN",
                        )
                        state.update(position_state_fields_from_attribution(orphan_attribution, action="OPEN"))
                        if not _position_owned_by_pipeline(state):
                            sync_open_fill(
                                orphan_attribution,
                                data_dir=DATA_DIR,
                                ticket=int(t or 0),
                                side=side_str,
                                volume=float(current_vol or 0.0),
                                entry_price=float(entry_price or 0.0),
                                sl=float(current_sl or 0.0),
                                tp=float(current_tp or 0.0),
                                symbol="XAUUSD",
                                source="bridge_reconcile_orphan",
                            )
                    except Exception as lifecycle_err:
                        _daemon_log(f"sync_open_fill orphan reconcile failed: {lifecycle_err}")
                    # 清理该 ticket 的确认计数
                    state.get("orphan_check_counts", {}).pop(orphan_key, None)
                    _log(f"Reconciliation: Recovered orphan position ticket={t} from MT5 (confirmed {orphan_check_count} times). state synchronized. position_time reset to now (orphan).", "WARN")

        conn.close()

        if persist_after_reconcile:
            # 在 DB 连接关闭后再持久化，避免锁竞争导致写入静默失败。
            save_state(state)

        # ── 5. 清理 DB 残留的 orphaned journal 文件 ──
        try:
            import glob as _glob
            journal_files = _glob.glob(str(DB_PATH) + "-mj-*") + _glob.glob(str(DB_PATH) + "-wal-*")
            for jf in journal_files:
                try:
                    os.unlink(jf)
                    _log(f"Reconciliation: Removed orphaned journal {jf}", "INFO")
                except OSError:
                    pass
        except Exception:
            pass

    except Exception as e:
        _log(f"Reconciliation failed: {e} | db={DB_PATH}", "ERROR")

    return state


# ============ CLI ============
if __name__ == "__main__":
    # [BLOCK] gold_trader.py daemon has been retired (2026-05-19).
    # The trading core is pipeline.py — run it directly:
    #   python3 src/pipeline.py --source ea_feed --runtime-intent live_trade ...
    # Utility functions (load_state, get_price, write_trade_request, etc.)
    # remain available as a library module.
    import sys as _sys
    _sys.exit(0)
