"""批量写入 .env 并刷新 Settings 缓存。""" from __future__ import annotations from pathlib import Path from .config import get_settings from .credentials import upsert_env_file def upsert_env_keys(updates: dict[str, str]) -> Path | None: """写入多项;空 value 跳过。返回最后写入的 .env 路径。""" target: Path | None = None for key, value in updates.items(): if value is None: continue # 允许显式清空密钥(传空串以外的 sentinel 由调用方决定);空串表示跳过 if value == "": continue target = upsert_env_file(key, value) get_settings.cache_clear() return target def mask_secret(raw: str | None, *, keep: int = 4) -> str | None: """脱敏:****末尾;过短则全部打码。""" s = (raw or "").strip() if not s: return None if len(s) <= keep: return "*" * len(s) return "*" * max(4, len(s) - keep) + s[-keep:] def okx_keys_configured(s=None) -> bool: st = s or get_settings() return bool( (st.okx_api_key or "").strip() and (st.okx_api_secret or "").strip() and (st.okx_api_passphrase or "").strip() ) def binance_keys_configured(s=None) -> bool: st = s or get_settings() return bool( (st.binance_api_key or "").strip() and (st.binance_api_secret or "").strip() ) def live_ready(*, exchange: str | None = None) -> tuple[bool, str]: """LIVE 是否可下单。返回 (ok, reason)。""" from .exchange.runtime import load_runtime_settings, normalize_exchange_name st = get_settings() if st.is_sim: return True, "sim" ex = normalize_exchange_name(exchange or load_runtime_settings().exchange) if ex == "binance": if not binance_keys_configured(st): return False, "币安 API Key/Secret 未配置" return True, "ok" if ex == "okx": if not okx_keys_configured(st): return False, "OKX API Key/Secret/Passphrase 未配置" return True, "ok" return False, f"未知交易所: {ex}"