from __future__ import annotations
import os
import re
import sys
import json
import pathlib
import logging
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Optional, Union

_log = logging.getLogger(__name__)
_APP_DIR_CACHE: Path | None = None

try:
    from src.atomic_write import atomic_write_json
except Exception:  # pragma: no cover - fallback for very early bootstrap paths
    atomic_write_json = None

_RUNTIME_TEMPLATE_DIRS = ("config", "data", "ops", "tenants")
_STALE_JSON_TEMP_PATTERNS = (
    re.compile(r".+\.tmp$", re.IGNORECASE),
    re.compile(r"^(?:app_config|path_cache|llm_providers|openclaw)_[^.].*\.json$", re.IGNORECASE),
    re.compile(r".+\.json_[^.].*\.json$", re.IGNORECASE),
)


def _coerce_int(value: Any) -> int | None:
    try:
        return int(float(value))
    except (TypeError, ValueError):
        return None


def _merge_runtime_config_dicts(
    *,
    filename: str,
    bundled: dict[str, Any],
    existing: dict[str, Any],
) -> dict[str, Any]:
    if filename == "llm_providers.json":
        merged = dict(bundled)
        merged.update(existing)
        bundled_providers = bundled.get("providers", {})
        existing_providers = existing.get("providers", {})
        merged_providers = dict(bundled_providers)
        merged_providers.update(existing_providers)
        merged["providers"] = merged_providers
        return merged

    merged = dict(bundled)
    merged.update(existing)

    if filename == "risk_governor_rules.json":
        for key in ("max_positions_per_strategy", "max_same_side_per_strategy"):
            bundled_limit = _coerce_int(bundled.get(key))
            existing_limit = _coerce_int(merged.get(key))
            if bundled_limit is None or bundled_limit <= 0:
                continue
            if existing_limit is None or existing_limit <= 0:
                merged[key] = bundled_limit
                continue
            # Guardrails may tighten across releases, but should not silently loosen
            # because an old runtime copy still carries a stale larger limit.
            merged[key] = min(existing_limit, bundled_limit)

    return merged


def _safe_copy2(src: Path, dst: Path) -> None:
    try:
        dst.parent.mkdir(parents=True, exist_ok=True)
        import shutil

        shutil.copy2(src, dst)
    except Exception:
        _log.warning("Config bootstrap skipped for %s", dst, exc_info=True)


def _safe_copytree(src: Path, dst: Path) -> None:
    try:
        import shutil

        shutil.copytree(src, dst, dirs_exist_ok=True)
    except Exception:
        _log.warning("Config bootstrap skipped for %s", dst, exc_info=True)


def _write_json_atomic(path: Path, payload: dict[str, Any], *, context: str) -> bool:
    if atomic_write_json is None:
        _log.error("%s: atomic_write_json unavailable for %s", context, path)
        return False
    try:
        ok = bool(atomic_write_json(path, payload))
    except Exception:
        _log.exception("%s: atomic write raised for %s", context, path)
        return False
    if not ok:
        _log.error("%s: atomic write failed for %s", context, path)
    return ok


def _copy_missing_tree(src: Path, dst: Path) -> None:
    if not src.exists():
        return
    for root, dirs, files in os.walk(src):
        src_root = Path(root)
        rel_root = src_root.relative_to(src)
        dst_root = dst / rel_root if str(rel_root) != "." else dst
        dst_root.mkdir(parents=True, exist_ok=True)
        for dirname in dirs:
            (dst_root / dirname).mkdir(parents=True, exist_ok=True)
        for filename in files:
            target = dst_root / filename
            if not target.exists():
                _safe_copy2(src_root / filename, target)


def _cleanup_stale_json_temp_files(base_dir: Path) -> None:
    cleaned = 0
    for rel_dir in ("config", "data", "ops", "tenants", "logs"):
        root = base_dir / rel_dir
        if not root.exists():
            continue
        try:
            for path in root.rglob("*"):
                if cleaned >= 200:
                    return
                if not path.is_file():
                    continue
                name = path.name
                if not any(pattern.match(name) for pattern in _STALE_JSON_TEMP_PATTERNS):
                    continue
                try:
                    path.unlink()
                    cleaned += 1
                except Exception:
                    _log.warning("Failed to remove stale temp file %s", path, exc_info=True)
        except Exception:
            _log.warning("Failed to scan temp files under %s", root, exc_info=True)


def _runtime_dir_writable(path: Path) -> bool:
    target = path if path.suffix == "" else path.parent
    try:
        target.mkdir(parents=True, exist_ok=True)
    except Exception:
        return False
    return os.access(target, os.W_OK)


def _managed_runtime_storage_root() -> Path | None:
    code_dir = Path(__file__).resolve().parent
    for parent in code_dir.parents:
        candidate = parent / "managed_app" / "releases"
        try:
            code_dir.relative_to(candidate)
        except ValueError:
            continue
        return parent
    return None


def _env_storage_root() -> Path | None:
    raw = (
        os.environ.get("GOLD_AUTOTRADE_APP_DIR", "").strip()
        or os.environ.get("GOLD_AUTOTRADE_STORAGE_ROOT", "").strip()
    )
    if not raw:
        return None
    try:
        return Path(raw).expanduser().resolve()
    except Exception:
        return Path(raw).expanduser()


def _get_storage_root() -> Path:
    env_root = _env_storage_root()
    if env_root is not None:
        env_root.mkdir(parents=True, exist_ok=True)
        return env_root
    managed_root = _managed_runtime_storage_root()
    if managed_root is not None:
        managed_root.mkdir(parents=True, exist_ok=True)
        return managed_root
    return get_app_dir()


# 获取应用级配置文件的位置 (放在项目根目录下，如果是开发模式)
# 如果是打包模式，它会放在用户数据目录下
def get_app_dir() -> Path:
    """获取程序运行时的根数据目录"""
    global _APP_DIR_CACHE
    if _APP_DIR_CACHE is not None:
        return _APP_DIR_CACHE
    env_root = _env_storage_root()
    if env_root is not None:
        env_root.mkdir(parents=True, exist_ok=True)
        _APP_DIR_CACHE = env_root
        return env_root
    # 检查是否是被 PyInstaller 打包后的环境
    if getattr(sys, 'frozen', False):
        # 生产环境：使用系统 APPDATA
        if sys.platform == "win32":
            base_dir = Path(os.getenv("APPDATA", "~")) / "GoldAutoTrade"
        else:
            # macOS 下，为了避免沙盒权限问题，直接把数据存在当前执行文件所在的同级目录下
            base_dir = Path(sys.executable).parent / "GoldAutoTradeData"

        bundled_root = Path(sys._MEIPASS)
        # [重要] 首次运行时，如果打包环境内置了 config 和 data，释放它们
        #        升级时，合并新增的配置键，不覆盖用户已有的自定义值
        if not base_dir.exists():
            base_dir.mkdir(parents=True, exist_ok=True)
            for dirname in _RUNTIME_TEMPLATE_DIRS:
                bundled_dir = bundled_root / dirname
                if bundled_dir.exists():
                    _safe_copytree(bundled_dir, base_dir / dirname)

            bundled_version = bundled_root / "VERSION"
            if bundled_version.exists():
                _safe_copy2(bundled_version, base_dir / "VERSION")

            bundled_app_config = bundled_root / "app_config.json"
            if bundled_app_config.exists():
                _safe_copy2(bundled_app_config, base_dir / "app_config.json")
        else:
            # 升级时：合并 bundled config 中新增的顶层键，不覆盖用户已有配置
            import shutil
            _cleanup_stale_json_temp_files(base_dir)
            # Always update VERSION to reflect the running build
            bundled_version = bundled_root / "VERSION"
            if bundled_version.exists():
                try:
                    shutil.copy2(bundled_version, base_dir / "VERSION")
                except Exception:
                    _log.warning("Failed to refresh VERSION at %s", base_dir / "VERSION", exc_info=True)
            # Seed default app_config.json if missing (e.g. fresh config dir without bootstrap)
            bundled_app_config = bundled_root / "app_config.json"
            if bundled_app_config.exists() and not (base_dir / "app_config.json").exists():
                try:
                    shutil.copy2(bundled_app_config, base_dir / "app_config.json")
                except Exception:
                    _log.warning("Failed to seed app_config.json into %s", base_dir, exc_info=True)
            bundled_config = bundled_root / "config"
            if bundled_config.exists():
                runtime_config_dir = base_dir / "config"
                if _runtime_dir_writable(runtime_config_dir):
                    for fname in bundled_config.iterdir():
                        # Copy subdirectories (e.g. personas/) that don't exist in runtime
                        if fname.is_dir() and not fname.name.startswith('.'):
                            target = runtime_config_dir / fname.name
                            if not target.exists():
                                _safe_copytree(fname, target)
                                _log.info("Config upgrade: copied new directory %s", fname.name)
                            # If target exists, merge individual files from bundled
                            else:
                                for child in fname.iterdir():
                                    if child.is_file() and child.suffix == '.json':
                                        child_target = target / child.name
                                        if not child_target.exists():
                                            _safe_copy2(child, child_target)
                                            _log.info("Config upgrade: copied new persona %s", child.name)
                            continue
                        if fname.suffix != ".json":
                            continue
                        target = runtime_config_dir / fname.name
                        target.parent.mkdir(parents=True, exist_ok=True)
                        if not target.exists():
                            _safe_copy2(fname, target)
                            continue
                        try:
                            existing = json.loads(target.read_text(encoding="utf-8"))
                            bundled = json.loads(fname.read_text(encoding="utf-8"))
                            if not isinstance(existing, dict) or not isinstance(bundled, dict):
                                continue
                            merged = _merge_runtime_config_dicts(
                                filename=fname.name,
                                bundled=bundled,
                                existing=existing,
                            )
                            _write_json_atomic(target, merged, context=f"config upgrade merge {fname.name}")
                        except Exception:
                            _log.warning("Config upgrade merge failed for %s", target, exc_info=True)
                else:
                    _log.warning("Config upgrade skipped: runtime config dir is not writable: %s", runtime_config_dir)
            for dirname in ("data", "ops", "tenants"):
                bundled_dir = bundled_root / dirname
                if bundled_dir.exists():
                    _copy_missing_tree(bundled_dir, base_dir / dirname)
        # 确保所有运行态子目录存在（首次 + 升级都执行）
        for _sub in ("logs", "data/bridge", "data/inbox", "data/outbox",
                     "data/auth", "data/compliance", "data/bridge/archive",
                     "data/learning", "data/bars_cache",
                     "memory/trades", "memory/strategies",
                     "memory/patterns", "memory/lessons", "memory/daily",
                     "memory/weekly", "memory/regimes", "memory/agents"):
            (base_dir / _sub).mkdir(parents=True, exist_ok=True)
    else:
        # 开发环境：使用当前项目根目录，保证你的现有开发状态不被破坏
        base_dir = Path(__file__).parent

    base_dir.mkdir(parents=True, exist_ok=True)
    _APP_DIR_CACHE = base_dir
    return base_dir


def get_project_dir() -> Path:
    """Return the project root directory (same as get_app_dir)."""
    return get_app_dir()


def get_app_config_path() -> Path:
    return _get_storage_root() / "app_config.json"


# ── Path cache (auto-discovered paths, separate from user config) ──────

def _path_cache_path() -> Path:
    return get_config_dir() / "path_cache.json"


def _load_path_cache() -> dict[str, str]:
    """Load auto-discovered path cache. Never fails — returns {} on any error."""
    p = _path_cache_path()
    if not p.exists():
        return {}
    try:
        data = json.loads(p.read_text(encoding="utf-8"))
        return data if isinstance(data, dict) else {}
    except Exception:
        return {}


def _save_path_cache(data: dict[str, str]) -> None:
    """Atomic write for path cache (auto-discovered paths only)."""
    p = _path_cache_path()
    p.parent.mkdir(parents=True, exist_ok=True)
    _write_json_atomic(p, data, context="save path cache")


# ── app_config.json read/write ──────────────────────────────────────

def load_app_config() -> dict:
    """Read app_config.json. Returns {} if file doesn't exist.

    If the file exists but cannot be parsed as valid JSON, the corrupted
    file is renamed to .corrupted and an error is logged — callers receive
    {} so the system can continue running, but the corruption is NOT
    silently swallowed.
    """
    config_path = get_app_config_path()
    if not config_path.exists():
        return {}

    raw = config_path.read_text(encoding="utf-8")
    if not raw.strip():
        return {}

    try:
        data = json.loads(raw)
    except json.JSONDecodeError:
        _log.error(
            "app_config.json is corrupted and cannot be parsed. "
            "Renaming to .corrupted and returning empty config. "
            "Restore from backup or reconfigure."
        )
        _backup_corrupted(config_path)
        return {}

    if not isinstance(data, dict):
        _log.error(
            "app_config.json is not a JSON object (type=%s). "
            "Renaming to .corrupted and returning empty config.",
            type(data).__name__,
        )
        _backup_corrupted(config_path)
        return {}

    return data


def _backup_corrupted(config_path: Path) -> None:
    """Rename corrupted config file so it can be inspected later."""
    try:
        corrupted_path = config_path.with_suffix(".corrupted")
        # If a .corrupted already exists, add timestamp
        if corrupted_path.exists():
            ts = datetime.now().strftime("%Y%m%d_%H%M%S")
            corrupted_path = config_path.with_suffix(f".corrupted.{ts}")
        config_path.rename(corrupted_path)
    except Exception:
        pass


def save_app_config(data: dict) -> None:
    """Atomically merge `data` into app_config.json.

    Reads current config, merges the new keys, writes to a temp file,
    then atomically replaces via os.replace(). A .bak backup is created
    before the replace.
    """
    config_path = get_app_config_path()
    config_path.parent.mkdir(parents=True, exist_ok=True)

    # Read current state (start fresh if corrupted/missing)
    try:
        current = load_app_config()
    except Exception:
        current = {}
    current.update(data)

    # Validate round-trip before writing
    try:
        payload = json.dumps(current, indent=2, ensure_ascii=False)
        json.loads(payload)  # verify valid JSON
    except Exception:
        _log.error("save_app_config: data cannot be serialized to JSON — aborting")
        return

    # Create .bak backup of existing file
    if config_path.exists():
        try:
            bak_path = config_path.with_suffix(".json.bak")
            bak_path.write_bytes(config_path.read_bytes())
        except Exception:
            pass

    try:
        parsed_payload = json.loads(payload)
    except Exception:
        _log.exception("save_app_config: payload parse-back failed")
        raise
    if not _write_json_atomic(config_path, parsed_payload, context="save app config"):
        raise RuntimeError(f"save_app_config failed for {config_path}")


def _scan_windows_mt5_dirs() -> dict[str, str]:
    """暴力扫描 Windows 下所有可能的 MT5 数据目录。

    返回:
        "Common/Files" — 找到的可用的 MT5 Common/Files 目录
        "instances" — 所有找到的 MT5 terminal 实例目录 (含子目录名)
        "bridge_dir" — 包含 gold_autotrade/bridge 的目录
    """
    result: dict[str, str] = {}
    instances: list[str] = []
    bridge_dir: str = ""

    # 优先收集：有 gold_autotrade/bridge 的 Common/Files 路径
    bridge_common_path: str = ""
    # 兜底：第一个找到的 Common/Files（无 bridge 数据）
    first_common_path: str = ""

    # ── 策略 1: 标准 APPDATA/MetaQuotes/Terminal/<instance>/Common/Files ──
    base = os.environ.get("APPDATA", "")
    if base:
        terminal_root = Path(base) / "MetaQuotes" / "Terminal"
        if terminal_root.exists():
            direct_common_files = terminal_root / "Common" / "Files"
            if direct_common_files.exists():
                path_str = str(direct_common_files.resolve())
                instances.append(path_str)
                if not first_common_path:
                    first_common_path = path_str
                bridge = direct_common_files / "gold_autotrade" / "bridge"
                if bridge.exists():
                    if not bridge_dir:
                        bridge_dir = str(bridge)
                    if not bridge_common_path:
                        bridge_common_path = path_str
            for inst_dir in sorted(terminal_root.iterdir()):
                if not inst_dir.is_dir():
                    continue
                common_files = inst_dir / "Common" / "Files"
                if common_files.exists():
                    path_str = str(common_files.resolve())
                    instances.append(path_str)
                    if not first_common_path:
                        first_common_path = path_str
                    bridge = common_files / "gold_autotrade" / "bridge"
                    if bridge.exists():
                        if not bridge_dir:
                            bridge_dir = str(bridge)
                        if not bridge_common_path:
                            bridge_common_path = path_str

    # ── 策略 2: 非标准品牌 MT5 (WeTrade, FTMO, etc.) ──
    brands_seen: set[str] = set()
    if base:
        appdata_path = Path(base)
        try:
            for entry in appdata_path.iterdir():
                if not entry.is_dir():
                    continue
                lower_name = entry.name.lower()
                if "metaquotes" in lower_name or "terminal" in lower_name:
                    continue
                if any(kw in lower_name for kw in ("metatrader", "wetrade", "ftmo", "trade")):
                    brands_seen.add(str(entry))
        except Exception:
            pass
        try:
            for entry in appdata_path.iterdir():
                lower_name = entry.name.lower()
                if "terminal" in lower_name and entry.is_dir() and str(entry) not in brands_seen:
                    brands_seen.add(str(entry))
        except Exception:
            pass

        for brand_path_str in brands_seen:
            brand_path = Path(brand_path_str)
            for inst_dir in sorted(brand_path.iterdir()) if brand_path.is_dir() else []:
                if not inst_dir.is_dir():
                    continue
                common_files = inst_dir / "Common" / "Files"
                if common_files and common_files.exists():
                    path_str = str(common_files.resolve())
                    if path_str not in instances:
                        instances.append(path_str)
                    if not first_common_path:
                        first_common_path = path_str
                    bridge = common_files / "gold_autotrade" / "bridge"
                    if bridge.exists():
                        if not bridge_dir:
                            bridge_dir = str(bridge)
                        if not bridge_common_path:
                            bridge_common_path = path_str

    # ── 策略 3: 注册表中读取 MT5 数据目录 ──
    try:
        import winreg
        for hive, subkey in [
            (winreg.HKEY_LOCAL_MACHINE, r"SOFTWARE\MetaQuotes\Terminal"),
            (winreg.HKEY_CURRENT_USER, r"SOFTWARE\MetaQuotes\Terminal"),
            (winreg.HKEY_LOCAL_MACHINE, r"SOFTWARE\WOW6432Node\MetaQuotes\Terminal"),
        ]:
            try:
                key = winreg.OpenKey(hive, subkey, 0, winreg.KEY_READ | winreg.KEY_WOW64_64KEY)
                try:
                    i = 0
                    while True:
                        inst_path_str = winreg.EnumKey(key, i)
                        i += 1
                        try:
                            inst_key = winreg.OpenKey(key, inst_path_str)
                            try:
                                data_dir_str, _ = winreg.QueryValueEx(inst_key, "DataFolder")
                                if data_dir_str and os.path.isdir(data_dir_str):
                                    common_files = Path(data_dir_str) / "Common" / "Files"
                                    if common_files.exists():
                                        path_str = str(common_files.resolve())
                                        if path_str not in instances:
                                            instances.append(path_str)
                                        if not first_common_path:
                                            first_common_path = path_str
                                        bridge = common_files / "gold_autotrade" / "bridge"
                                        if bridge.exists():
                                            if not bridge_dir:
                                                bridge_dir = str(bridge)
                                            if not bridge_common_path:
                                                bridge_common_path = path_str
                            except Exception:
                                pass
                            try:
                                inst_key.Close()
                            except Exception:
                                pass
                        except Exception:
                            pass
                except Exception:
                    pass
                try:
                    key.Close()
                except Exception:
                    pass
            except Exception:
                pass
    except Exception:
        pass

    # ── 策略 4: 扫描 program files 目录下的 MQL5/Files ──
    # 注意: EA 使用 FILE_COMMON 写入 Common/Files，MQL5/Files 是本地文件路径，
    # 仅作为 instances 收集，不设置为 Common/Files 结果，避免路径误匹配。
    program_files = [
        os.environ.get("ProgramFiles", "C:/Program Files"),
        os.environ.get("ProgramFiles(x86)", "C:/Program Files (x86)"),
    ]
    mql5_bridge_dir: str = ""
    for pf in program_files:
        if not pf:
            continue
        try:
            pf_path = Path(pf)
            if not pf_path.exists():
                continue
            for mt5_dir_entry in pf_path.iterdir():
                if not mt5_dir_entry.is_dir():
                    continue
                lower = mt5_dir_entry.name.lower()
                if any(kw in lower for kw in ("metatrader", "wetrade", "ftmo")):
                    mql5_files = mt5_dir_entry / "MQL5" / "Files"
                    if mql5_files.exists():
                        path_str = str(mql5_files.resolve())
                        if path_str not in instances:
                            instances.append(path_str)
                        bridge = mql5_files / "gold_autotrade" / "bridge"
                        if bridge.exists() and not bridge_dir:
                            bridge_dir = str(bridge)
                            mql5_bridge_dir = path_str
        except Exception:
            pass

    # ── 结果合并: 优先有 bridge 数据的 Common/Files ──
    if bridge_common_path:
        result["Common/Files"] = bridge_common_path
    elif first_common_path:
        result["Common/Files"] = first_common_path
    elif mql5_bridge_dir:
        # 仅在完全没有 Common/Files 但有 MQL5 bridge 时回退
        result["Common/Files"] = mql5_bridge_dir

    # ── 策略 5: 全盘搜索 gold_autotrade/bridge 目录（最终兜底） ──
    # 【修复】原代码使用 /mnt/c 等 WSL 路径，在 Windows 原生环境下无效。
    # 现在区分 Windows 原生 vs WSL/Linux 探测。
    if not bridge_dir and sys.platform == "win32":
        # 5a: Windows 原生 — 扫描已知盘符根目录下 Program Files / ProgramData
        import string
        scanned_dirs: set[str] = set()
        for env_key in ("ProgramFiles", "ProgramFiles(x86)", "LOCALAPPDATA", "ProgramData"):
            val = os.environ.get(env_key, "").strip()
            if val:
                scanned_dirs.add(val)
        # 补充 C: D: E: … H: 的 Program Files 变体
        for letter in string.ascii_uppercase[:8]:
            for sub in ("Program Files", "Program Files (x86)", "ProgramData"):
                p = f"{letter}:\\{sub}"
                if os.path.isdir(p):
                    scanned_dirs.add(p)
        for root_dir in scanned_dirs:
            try:
                for entry in os.listdir(root_dir):
                    lower = entry.lower()
                    if any(kw in lower for kw in ("metatrader", "terminal64", "ftmo", "wetrade")):
                        full = Path(root_dir) / entry
                        if not full.is_dir():
                            continue
                        mql5_files = full / "MQL5" / "Files"
                        if mql5_files.exists():
                            path_str = str(mql5_files.resolve())
                            if path_str not in instances:
                                instances.append(path_str)
                            if not result.get("Common/Files"):
                                result["Common/Files"] = path_str
                            bridge = mql5_files / "gold_autotrade" / "bridge"
                            if bridge.exists() and not bridge_dir:
                                bridge_dir = str(bridge)
                                result["bridge_dir"] = str(bridge)
            except Exception:
                continue

    # 5b/6: 通过 psutil 从运行中的 MT5 进程反推安装目录（Windows + macOS 通用）
    if not result.get("Common/Files"):
        try:
            import psutil
            terminal_names = ("terminal64.exe", "terminal64", "terminal.exe", "terminal")
            for proc in psutil.process_iter(["name", "exe"]):
                p_name = str(proc.info.get("name", "")).lower().strip()
                if p_name not in terminal_names:
                    continue
                exe_path = str(proc.info.get("exe", "") or "").strip()
                if not exe_path or not Path(exe_path).exists():
                    continue
                exe_dir = Path(exe_path).parent
                # MT5 安装目录下可能有 MQL5/Files 或 Common/Files
                candidates = [
                    exe_dir / "MQL5" / "Files",
                    exe_dir.parent / "MQL5" / "Files",
                ]
                for cf in candidates:
                    if cf.exists():
                        path_str = str(cf.resolve())
                        if path_str not in instances:
                            instances.append(path_str)
                        if not result.get("Common/Files"):
                            result["Common/Files"] = path_str
                        bridge = cf / "gold_autotrade" / "bridge"
                        if bridge.exists() and not bridge_dir:
                            bridge_dir = str(bridge)
                            result["bridge_dir"] = str(bridge)
                        break
                if result.get("Common/Files"):
                    break
        except Exception:
            pass

    # 5c: WSL/Linux/macOS Wine 路径（原逻辑保留，仅在非 Windows 平台执行）
    if not result.get("Common/Files") and sys.platform != "win32":
        for drive_letter in ("C:", "D:", "E:"):
            drive_root = Path(f"/mnt/{drive_letter[0].lower()}")
            if not drive_root.exists():
                continue
            try:
                for candidate in drive_root.rglob("gold_autotrade/bridge"):
                    if candidate.is_dir():
                        bridge_dir = str(candidate)
                        result["bridge_dir"] = bridge_dir
                        if not result.get("Common/Files"):
                            result["Common/Files"] = str(candidate.parent.parent)
                        break
            except Exception:
                continue
            if bridge_dir:
                break

    result["instances"] = instances
    if bridge_dir:
        result["bridge_dir"] = bridge_dir
    return result


def _has_live_mt5_bridge_data(p: Path) -> bool:
    try:
        status_candidates = [
            p / "gold_autotrade" / "bridge" / "terminal_status.json",
            p / "gold_autotrade" / "bridge" / "from_ea" / "terminal_status.json",
            p / "gold_autotrade" / "bridge" / "market_snapshot.json",
            p / "gold_autotrade" / "bridge" / "from_ea" / "market_snapshot.json",
        ]
        if any(candidate.exists() for candidate in status_candidates):
            return True
        return False
    except Exception:
        return False


def get_mt5_dir():
    """获取MT5的数据目录，优先读用户配置，没有则自动猜测"""
    config = load_app_config()
    fallback_existing: Path | None = None

    # 1. 优先读取用户在 app_config.json 中的手动配置
    if config.get("mt5_data_dir"):
        p = Path(config["mt5_data_dir"])
        if p.exists():
            if _has_live_mt5_bridge_data(p):
                return p
            fallback_existing = p

    # 2. 读取自动发现的路径缓存 (path_cache.json)
    path_cache = _load_path_cache()
    if path_cache.get("mt5_data_dir"):
        p = Path(path_cache["mt5_data_dir"])
        if p.exists():
            if _has_live_mt5_bridge_data(p):
                return p
            if fallback_existing is None:
                fallback_existing = p

    # 3. 动态探测 (绝不使用硬编码的用户名)
    if sys.platform == "darwin":
        import getpass
        # 探测 Wine 容器基础路径
        wine_base = Path.home() / "Library/Application Support/net.metaquotes.wine.metatrader5/drive_c/users"
        if wine_base.exists():
            # 自动遍历 Wine 容器内的用户目录 (可能是 user, 可能是当前系统用户名)
            potential_users = ["user", getpass.getuser()]
            try:
                potential_users.extend([d.name for d in wine_base.iterdir() if d.is_dir()])
            except Exception:
                pass

            for u in set(potential_users):
                p = wine_base / u / "AppData/Roaming/MetaQuotes/Terminal/Common/Files"
                if p.exists():
                    # 自动发现的路径写入 path_cache，不动 app_config
                    path_cache["mt5_data_dir"] = str(p)
                    _save_path_cache(path_cache)
                    return p

    elif sys.platform == "win32":
        # Windows 原生探测 — 全面嗅探所有可能的 MT5 数据目录
        scanned = _scan_windows_mt5_dirs()

        # 优先：有 gold_autotrade/bridge 的 Common/Files
        bridge_dir = scanned.get("bridge_dir")
        if bridge_dir:
            bridge_common = Path(bridge_dir).parent.parent  # .../Common/Files/gold_autotrade/bridge → .../Common/Files
            if bridge_common.exists():
                path_cache["mt5_data_dir"] = str(bridge_common)
                _save_path_cache(path_cache)
                return bridge_common

        # 次优：扫描到的 Common/Files
        mt5_data_dir = scanned.get("Common/Files")
        if mt5_data_dir and Path(mt5_data_dir).exists():
            p = Path(mt5_data_dir)
            path_cache["mt5_data_dir"] = str(p)
            _save_path_cache(path_cache)
            return p

        # 兜底：通过 bridge_dir 反推 Common/Files
        if bridge_dir:
            p = Path(bridge_dir).parent  # gold_autotrade/bridge 的父级（即 gold_autotrade/）
            path_cache["mt5_data_dir"] = str(p)
            _save_path_cache(path_cache)
            return p

    if fallback_existing is not None:
        return fallback_existing

    # 4. 最终回退：项目相对路径 (用于 Portable 模式)
    fallback_p = get_app_dir().parent / "bridge/Files"
    if fallback_p.exists():
        return fallback_p

    # 如果实在找不到，返回一个安全的默认占位符，避免抛出异常导致整个系统无法启动
    return get_app_dir().parent / "bridge/Files"


def get_llm_config_dir() -> Path:
    """获取大模型配置文件所在目录 (openclaw.json)

    优先读 app_config.json 用户手动设置的路径，其次读 path_cache.json
    自动发现的路径，最后 fallback 到 ~/.openclaw。
    不再将自动发现的路径写入 app_config.json（避免读函数产生写副作用）。
    """
    config = load_app_config()

    # 1. 用户在 app_config.json 中手动设置的路径
    if "llm_config_dir" in config and config["llm_config_dir"]:
        cached = Path(config["llm_config_dir"])
        if cached.exists():
            return cached
        # cached path is stale (e.g. deployed to a different machine);
        # fall through and recompute

    # 2. 读取自动发现的路径缓存 (path_cache.json)
    path_cache = _load_path_cache()
    if path_cache.get("llm_config_dir"):
        p = Path(path_cache["llm_config_dir"])
        if p.exists():
            return p

    # 3. 默认 fallback 到 ~/.openclaw（只缓存到 path_cache，不写 app_config）
    default_dir = Path.home() / ".openclaw"
    if default_dir.exists():
        path_cache["llm_config_dir"] = str(default_dir)
        _save_path_cache(path_cache)
        return default_dir

    return default_dir


def save_provider_api_key(provider: str, api_key: str) -> bool:
    """Save API key for a provider to llm_providers.json (single source of truth).

    llm_providers.json is the authoritative key store. Legacy openclaw.json
    is updated as a side-effect for backward compatibility only.
    """
    # Primary: write to llm_providers.json (authoritative source)
    config_dir = get_config_dir()
    llm_path = config_dir / "llm_providers.json"
    saved_primary = False
    try:
        cfg = {}
        if llm_path.exists():
            cfg = json.loads(llm_path.read_text(encoding="utf-8"))
        cfg.setdefault("providers", {})
        cfg["providers"].setdefault(provider, {})
        cfg["providers"][provider]["apiKey"] = api_key

        payload = json.dumps(cfg, indent=2, ensure_ascii=False)

        if llm_path.exists():
            try:
                bak_path = llm_path.with_suffix(".json.bak")
                bak_path.write_bytes(llm_path.read_bytes())
            except Exception:
                _log.warning("Failed to refresh llm_providers backup %s", llm_path, exc_info=True)
        saved_primary = _write_json_atomic(llm_path, cfg, context="save llm_providers")
    except Exception:
        _log.exception("Failed to save llm_providers.json")

    # Secondary: update legacy openclaw.json for backward compat (consumer code
    # that still reads it directly)
    try:
        openclaw_path = get_llm_config_dir() / "openclaw.json"
        oc_cfg = {}
        if openclaw_path.exists():
            oc_cfg = json.loads(openclaw_path.read_text(encoding="utf-8"))
        oc_cfg.setdefault("models", {}).setdefault("providers", {})
        oc_cfg["models"]["providers"].setdefault(provider, {})
        oc_cfg["models"]["providers"][provider]["apiKey"] = api_key
        oc_payload = json.dumps(oc_cfg, indent=2, ensure_ascii=False)
        if openclaw_path.exists():
            try:
                bak_path = openclaw_path.with_suffix(".json.bak")
                bak_path.write_bytes(openclaw_path.read_bytes())
            except Exception:
                _log.warning("Failed to refresh openclaw backup %s", openclaw_path, exc_info=True)
        openclaw_path.parent.mkdir(parents=True, exist_ok=True)
        _write_json_atomic(openclaw_path, oc_cfg, context="save legacy openclaw config")
    except Exception:
        _log.exception("Failed to update legacy openclaw.json")

    return saved_primary


def get_logs_dir() -> Path:
    d = _get_storage_root() / "logs"
    d.mkdir(exist_ok=True)
    return d

def get_data_dir() -> Path:
    d = _get_storage_root() / "data"
    d.mkdir(exist_ok=True)
    return d

def get_config_dir() -> Path:
    d = _get_storage_root() / "config"
    d.mkdir(exist_ok=True)
    return d


def _load_json_dict(path: Path) -> dict[str, Any]:
    if not path.exists():
        return {}
    import time as _time
    last_err = None
    for _ in range(5):
        try:
            payload = json.loads(path.read_text(encoding="utf-8-sig"))
            return payload if isinstance(payload, dict) else {}
        except (json.JSONDecodeError, PermissionError, OSError) as e:
            last_err = e
            _time.sleep(0.3)
    return {}


def _runtime_strategy_rules_path() -> Path:
    runtime_file = get_config_dir() / "strategy_rules_v3.json"
    if runtime_file.exists():
        return runtime_file
    repo_file = Path(__file__).resolve().parent / "config" / "strategy_rules_v3.json"
    if repo_file.exists():
        return repo_file
    return runtime_file


def baseline_config_path() -> Path:
    return _runtime_strategy_rules_path()


def override_config_path() -> Path:
    return get_config_dir() / "risk_governor_rules.json"


def effective_config_path() -> Path:
    return get_logs_dir() / "effective_params.json"


def load_baseline_config() -> dict[str, Any]:
    return _load_json_dict(baseline_config_path())


def load_override_config() -> dict[str, Any]:
    return _load_json_dict(override_config_path())


def load_effective_runtime_config() -> dict[str, Any]:
    return normalize_effective_runtime_payload(_load_json_dict(effective_config_path()))


def _normalize_effective_suggestion(raw: Any) -> dict[str, Any] | None:
    if not isinstance(raw, dict):
        return None
    param_name = str(raw.get("param", "") or "").strip()
    if not param_name:
        return None
    value = raw.get("applied_value")
    if value is None:
        value = raw.get("suggested_value")
    if value is None:
        value = raw.get("value")
    normalized = dict(raw)
    normalized.setdefault("applied_value", value)
    normalized.setdefault("suggested_value", value)
    normalized.setdefault("param", param_name)
    return normalized


def normalize_effective_runtime_payload(payload: Any) -> dict[str, Any]:
    """Normalize multiple effective_params schemas into one compatible payload.

    Supported sources:
    - legacy review schema: {"suggestions": [...]}
    - training/server schema: {"current": {"params": ...}, "history": [...]}
    - runtime schema: {"params": {...}}
    """
    if not isinstance(payload, dict):
        return {}

    normalized = dict(payload)
    params: dict[str, Any] = {}
    suggestions: list[dict[str, Any]] = []
    seen_params: set[str] = set()

    top_level_params = payload.get("params", {})
    if isinstance(top_level_params, dict):
        params.update(top_level_params)

    offline_params = payload.get("offline_params", {})
    if isinstance(offline_params, dict):
        for key, value in offline_params.items():
            if str(key).startswith("_") or key in {"review_pending"}:
                continue
            params[str(key)] = value

    current = payload.get("current", {})
    if isinstance(current, dict):
        current_params = current.get("params", {})
        if isinstance(current_params, dict):
            params.update(current_params)
        normalized.setdefault(
            "updated_at",
            current.get("synced_at") or current.get("applied_at") or current.get("generated_at", ""),
        )
        normalized.setdefault("source", current.get("source", payload.get("source", "")))

    raw_suggestions = payload.get("suggestions", [])
    if isinstance(raw_suggestions, list):
        for item in raw_suggestions:
            suggestion = _normalize_effective_suggestion(item)
            if suggestion is None:
                continue
            suggestions.append(suggestion)
            seen_params.add(str(suggestion["param"]))
            if suggestion.get("applied_value") is not None:
                params[str(suggestion["param"])] = suggestion.get("applied_value")

    if isinstance(current, dict) and isinstance(current.get("params"), dict):
        auto_apply = bool(current.get("auto_apply", False))
        applied = bool(current.get("applied", auto_apply))
        for param_name, value in current["params"].items():
            param_name = str(param_name)
            if param_name in seen_params:
                continue
            suggestions.append(
                {
                    "id": str(current.get("method", "effective_current") or "effective_current").upper(),
                    "param": param_name,
                    "applied_value": value,
                    "suggested_value": value,
                    "applied": applied,
                    "auto_apply": auto_apply,
                    "source": str(current.get("source", payload.get("source", "effective_current")) or "effective_current"),
                }
            )
            seen_params.add(param_name)

    if isinstance(payload.get("params"), dict):
        for param_name, value in payload["params"].items():
            param_name = str(param_name)
            if param_name in seen_params:
                continue
            suggestions.append(
                {
                    "id": "EFFECTIVE_PARAMS",
                    "param": param_name,
                    "applied_value": value,
                    "suggested_value": value,
                    "applied": True,
                    "auto_apply": False,
                    "source": str(payload.get("source", "effective_params") or "effective_params"),
                }
            )
            seen_params.add(param_name)

    normalized["params"] = params
    normalized["suggestions"] = suggestions
    if "updated_at" not in normalized:
        normalized["updated_at"] = str(payload.get("timestamp", "") or payload.get("generated_at", "") or "")
    return normalized


def _lookup_dotted_value(payload: dict[str, Any], dotted_key: str) -> Any:
    if dotted_key in payload:
        return payload[dotted_key]
    cursor: Any = payload
    for part in dotted_key.split("."):
        if not isinstance(cursor, dict) or part not in cursor:
            return None
        cursor = cursor[part]
    return cursor


def _candidate_param_keys(param_name: str) -> list[str]:
    alias_map = {
        "base_lots": ["position_sizing.base_lots"],
        "max_open_lots": ["position_sizing.max_lots", "max_position_size"],
        "max_position_size": ["position_sizing.max_lots"],
        "max_positions": ["risk_governor.max_positions"],
        "max_open_positions": ["risk_governor.max_positions", "max_positions"],
        "post_close_cooldown_sec": ["risk_governor.post_close_cooldown_sec"],
        "min_confidence": ["risk_governor.min_confidence", "min_confidence_threshold"],
        "min_confidence_threshold": ["risk_governor.min_confidence", "min_confidence"],
        "sl_pips": ["risk_governor.sl_pips"],
        "tp_pips": ["risk_governor.tp_pips"],
        "max_consecutive_losses": ["risk_governor.max_consecutive_losses"],
        "spread_kill": ["risk_governor.spread_kill"],
        "kelly_enabled": ["position_sizing.kelly_enabled"],
        "kelly_fraction": ["position_sizing.kelly_fraction"],
        "kelly_min_samples": ["position_sizing.kelly_min_samples"],
        "kelly_max_fraction": ["position_sizing.kelly_max_fraction"],
        "kelly_loss_streak_discount": ["position_sizing.kelly_loss_streak_discount"],
    }
    candidates: list[str] = [param_name]
    for alias in alias_map.get(param_name, []):
        if alias not in candidates:
            candidates.append(alias)
    if param_name.endswith("_mult_trend"):
        legacy_name = param_name.replace("_mult_trend", "")
        if legacy_name not in candidates:
            candidates.append(legacy_name)
    return candidates


def _get_schema_default(param_name: str) -> Any:
    """Read a parameter's default value from strategy_params_schema.py (single source of truth)."""
    try:
        from src.strategy_params_schema import get_schema_defaults_flat as _schema_flat

        return _schema_flat().get(param_name)
    except Exception:
        return None


def _resolve_effective_param(param_name: str, default: Any = None) -> tuple[Any, str]:
    candidate_keys = _candidate_param_keys(param_name)

    override_data = load_override_config()
    for key in candidate_keys:
        if key in override_data:
            return override_data[key], "override_config"

    effective_data = load_effective_runtime_config()
    effective_params = effective_data.get("params", {}) if isinstance(effective_data.get("params", {}), dict) else {}
    for key in candidate_keys:
        if key in effective_params:
            return effective_params[key], "effective_runtime"
    suggestions = effective_data.get("suggestions", []) if isinstance(effective_data.get("suggestions", []), list) else []
    for sug in suggestions:
        if not isinstance(sug, dict):
            continue
        if sug.get("param") in candidate_keys:
            return sug.get("applied_value", default), "effective_runtime"

    strategy_rules = load_baseline_config()
    for key in candidate_keys:
        value = _lookup_dotted_value(strategy_rules, key)
        if value is not None:
            return value, "baseline_config"

    for key in candidate_keys:
        schema_default = _get_schema_default(key)
        if schema_default is not None:
            return schema_default, "schema_default"

    return default, "caller_default"


def get_effective_param(param_name: str, default: Any = None):
    """
    全局统一参数获取入口。
    优先级：
    1. override 配置 (risk_governor_rules.json)
    2. effective 运行时快照 (effective_params.json)
    3. baseline 策略基线 (strategy_rules_v3.json)
    4. strategy_params_schema.py 中的 schema 默认值 (_schema_defaults)
    5. 调用者传入的 default 回退
    """
    value, _source = _resolve_effective_param(param_name, default)
    return value


def get_effective_param_with_source(param_name: str, default: Any = None) -> tuple[Any, str]:
    """Return both effective value and the source that supplied it."""
    return _resolve_effective_param(param_name, default)


def get_effective_min_confidence_threshold(default: float = 0.70) -> float:
    """统一入口：返回归一化后的最小置信度阈值 (0.0-1.0)。

    自动处理 GUI 面板输入的百分制 (0-100) → 小数制 (0-1) 转换。
    gold_trader / gui_backend / pipeline 共用此入口，消除硬编码分歧。
    """
    value, _source = _resolve_effective_param("min_confidence_threshold", default)
    raw = float(value) if value is not None else float(default)
    if raw > 1.0:
        raw = raw / 100.0
    return float(max(0.50, min(0.95, raw)))


def get_execution_cooldown_params() -> dict[str, Any]:
    """返回执行层冷却参数，从 risk_governor_rules.json execution.cooldown 段读取。

    pipeline.py 模块加载时调用，替代之前的硬编码常量。
    """
    gui_data = load_override_config()
    exec_section = gui_data.get("execution", {}) if isinstance(gui_data.get("execution"), dict) else {}
    cooldown = exec_section.get("cooldown", {}) if isinstance(exec_section, dict) else {}
    if not isinstance(cooldown, dict):
        cooldown = {}

    return {
        "ack_retry_cooldown_sec": float(cooldown.get("ack_retry_sec", 20.0) or 20.0),
        "timeout_retry_cooldown_sec": float(cooldown.get("timeout_retry_sec", 30.0) or 30.0),
        "pending_entry_open_limit_sec": int(cooldown.get("pending_entry_open_limit_sec", 1800) or 1800),
        "pending_entry_arm_wait_sec": int(cooldown.get("pending_entry_arm_wait_sec", 1200) or 1200),
        "max_timeout_retries": int(cooldown.get("max_timeout_retries", 3) or 3),
    }


def get_falling_knife_thresholds() -> dict[str, float]:
    """返回飞刀检测阈值，从 risk_governor_rules.json falling_knife 段读取。"""
    gui_data = load_override_config()
    fk = gui_data.get("falling_knife", {}) if isinstance(gui_data.get("falling_knife"), dict) else {}
    if not isinstance(fk, dict):
        fk = {}
    return {
        "high_m3": float(fk.get("high_m3_pct", 0.25) or 0.25),
        "high_weighted": float(fk.get("high_weighted_pct", 0.35) or 0.35),
        "medium_m3": float(fk.get("medium_m3_pct", 0.12) or 0.12),
        "medium_weighted": float(fk.get("medium_weighted_pct", 0.18) or 0.18),
    }


def get_interpreter_fallback_confidence() -> float:
    """返回 market_interpreter 未就绪时的回退置信度。"""
    gui_data = load_override_config()
    scoring = gui_data.get("scoring", {}) if isinstance(gui_data.get("scoring"), dict) else {}
    fallback = scoring.get("interpreter_fallback_confidence", None) if isinstance(scoring, dict) else None
    return float(fallback if fallback is not None else 0.55)


def get_macro_authority_thresholds() -> dict[str, float]:
    """返回宏观权威度阈值，从 risk_governor_rules.json macro_authority 段读取。"""
    gui_data = load_override_config()
    ma = gui_data.get("macro_authority", {}) if isinstance(gui_data.get("macro_authority"), dict) else {}
    if not isinstance(ma, dict):
        ma = {}
    return {
        "high": float(ma.get("high", 0.70) or 0.70),
        "medium": float(ma.get("medium", 0.45) or 0.45),
        "low": float(ma.get("low", 0.25) or 0.25),
    }


def get_scoring_floors() -> dict[str, dict[str, float]]:
    """返回策略评分的 conf/struct 地板值，从 risk_governor_rules.json scoring 段读取。"""
    gui_data = load_override_config()
    scoring = gui_data.get("scoring", {}) if isinstance(gui_data.get("scoring"), dict) else {}
    if not isinstance(scoring, dict):
        scoring = {}
    return {
        "trending": {
            "conf_floor": float(scoring.get("conf_floor_trending", 0.70) or 0.70),
            "struct_floor": float(scoring.get("struct_floor_trending", 0.65) or 0.65),
        },
        "ranging": {
            "conf_floor": float(scoring.get("conf_floor_ranging", 0.64) or 0.64),
            "struct_floor": float(scoring.get("struct_floor_ranging", 0.55) or 0.55),
        },
    }


def normalize_min_confidence_threshold(value: Any) -> float:
    """将置信度值归一化到 0.0-1.0 并裁切到 [0.50, 0.95]。

    GUI 面板可能传入百分制 (0-100)，这里自动转换。
    """
    raw = float(value) if value is not None else 0.70
    if raw > 1.0:
        raw = raw / 100.0
    return float(max(0.50, min(0.95, raw)))


def get_effective_params_snapshot(param_defaults: Optional[dict[str, Any]] = None) -> dict[str, Any]:
    """Build a compact snapshot of the values that live execution actually reads."""
    tracked_defaults: dict[str, Any] = {
        "risk_level": "balanced",
        "max_position_size": 0.10,
        "max_positions": 2,
        "sl_pips": 40,
        "tp_pips": 80,
        "human_in_the_loop": False,
        "min_confidence_threshold": 0.70,
        "account_total_loss_stop_pct": 30.0,
        "base_lots": 0.05,
        "max_open_lots": 0.10,
        "post_close_cooldown_sec": 300,
    }
    if isinstance(param_defaults, dict):
        tracked_defaults.update(param_defaults)

    snapshot: dict[str, Any] = {
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "sources": {},
    }
    for param_name, fallback in tracked_defaults.items():
        value, source = _resolve_effective_param(param_name, fallback)
        snapshot[param_name] = value
        snapshot["sources"][param_name] = source
    return snapshot
