"""env 配置页 UI 白名单:中文标签,按交易所过滤.""" from __future__ import annotations import os from typing import Any, Optional from lib.env.env_file_lib import env_get_all, read_env_lines from lib.env.env_schema import ( _field_type, _hot_reload, _is_sensitive, _mask_value, _restart_required, normalize_select_value, parse_env_example_schema, select_options_for, ) # 各所「交易所与实盘」字段(顺序即页面顺序) # 合约计仓/杠杆/自动单等已从 env UI 移除;永续划转等见系统设置开关(默认关) _EXCHANGE_LIVE_FIELDS: dict[str, list[tuple[str, str, str]]] = { "okx": [ ("LIVE_TRADING_ENABLED", "开启实盘下单", "关闭时仅走本地流程,不向交易所发单(期权共用)"), ("OKX_API_KEY", "API Key", "账户 API(期权共用)"), ("OKX_API_SECRET", "API Secret", "账户 API(期权共用)"), ("OKX_API_PASSPHRASE", "API Passphrase", "OKX 必填"), ("OKX_ACCOUNT_LABEL", "账户备注", "企业微信推送中显示"), ( "OKX_SHOW_PERP_FUNDS", "显示USDT资金/交易账户", "默认关闭;开启后顶栏显示 USDT 资金与交易账户。币本位期权买币仍会显示交易账户可用额", ), ], "binance": [ ("LIVE_TRADING_ENABLED", "开启实盘下单", "关闭时仅走本地流程,不向交易所发单"), ("BINANCE_API_KEY", "API Key", "永续子账户"), ("BINANCE_API_SECRET", "API Secret", "永续子账户"), ("BINANCE_MARGIN_MODE", "保证金模式", ""), ("BINANCE_POSITION_MODE", "持仓模式", ""), ("BINANCE_ACCOUNT_LABEL", "账户备注", "企业微信推送中显示"), ], "gate": [ ("LIVE_TRADING_ENABLED", "开启实盘下单", "关闭时仅走本地流程,不向交易所发单"), ("GATE_API_KEY", "API Key", "永续子账户"), ("GATE_API_SECRET", "API Secret", "永续子账户"), ("GATE_TD_MODE", "保证金模式", ""), ("GATE_POS_MODE", "持仓模式", ""), ("GATE_ACCOUNT_LABEL", "账户备注", "企业微信推送中显示"), ], } # 共享区块(企业微信等);对冲配置已从 UI 移除 _SHARED_SECTIONS: list[dict[str, Any]] = [ { "title": "企业微信", "fields": [ ("WECHAT_WEBHOOK", "机器人 Webhook", "行情与风控推送地址"), ("WECHAT_TIMEOUT_SECONDS", "推送超时(秒)", "默认 10"), ], }, { "title": "中控对接", "fields": [ ( "HUB_BRIDGE_TOKEN", "中控 Bridge Token", "首次打开本页自动生成;复制到 crypto_okx_hub「添加实例」即可;也可手动改后保存", ), ( "HUB_EMBED_FRAME_ANCESTORS", "允许嵌入的中控 Origin", "选填。一般可留空;仅当中控 iframe 打不开时再填中控地址(空格分隔),或调试填 *", ), ], }, ] _OPTIONS_SECTION: dict[str, Any] = { "title": "期权账户", "exchanges": frozenset({"okx"}), "fields": [ ("OKX_OPTIONS_ENABLED", "启用期权模块", "与上方 OKX_API_* 同源;不再单独配置期权密钥"), ("OKX_OPTIONS_ACCOUNT_LABEL", "期权账户备注", ""), ("OKX_OPTIONS_BUDGET_BUFFER", "预算缓冲比例", "如 0.95;币本位复利/固定预算共用"), ( "OKX_OPTIONS_COIN_COMPOUND", "按交易户USDT复利", "默认 true;预算=交易账户USDT×缓冲;关闭则用下方固定 USDT 预算×缓冲", ), ( "OKX_OPTIONS_COIN_BUDGET_USDT", "固定预算(USDT)", "仅复利关闭时生效", ), ( "OKX_OPTIONS_COIN_MAX_USDT_ENABLED", "单笔上限开关", "默认 false=靠人工转走控规模;true 时预算不超过下方 N U", ), ( "OKX_OPTIONS_COIN_MAX_USDT", "单笔上限(USDT)", "仅上限开关开启时生效", ), ( "OKX_OPTIONS_COIN_SPOT_BUY_BUFFER", "现货买入缓冲", "相对权利金倍数,默认 1.10(=多买10%);也可写 0.10 表示+10%。按最大可开张数×卖一权利金×本缓冲买币,不全额兑换", ), ( "OKX_OPTIONS_MAX_ACTIVE_POSITIONS", "期权持仓上限(笔)", "默认 0=不限制;按交易所期权合约笔数计数,同合约加仓不占新笔数", ), ("OKX_OPTIONS_DEFAULT_UNDERLY", "默认标的", "如 ETH"), ( "OKX_OPTIONS_CHAIN_MAX_DTE_DAYS", "期权链展示天数", "默认 14;下拉到期日只出现该天数内的合约(含明天)", ), ( "OKX_OPTIONS_MAX_DTE_DAYS", "开仓最大剩余天数", "默认 2;开仓时拒绝更远到期(与链展示天数独立)", ), ( "OKX_OPTIONS_CHAIN_ASK_LIQ_FILTER_ENABLED", "链上仅显示有卖一", "默认 true;开启后隐藏无卖一深度或深度不足1张的合约(含标记价估算行)", ), ( "OKX_OPTIONS_CLOSE_GATE_MODE", "目标平仓门控模式", "premium=可回收≥权利金×倍数(默认×1.05);net_pnl=净盈亏折USDT≥阈值", ), ( "OKX_OPTIONS_CLOSE_RECYCLE_MULT_COIN", "门控权利金倍数", "仅 premium 模式;默认 1.05", ), ( "OKX_OPTIONS_CLOSE_NET_PNL_MIN_USDT", "门控净盈亏下限(U)", "仅 net_pnl 模式;买一可回收−权利金折 USDT 后需≥此值;0 或 1 等", ), ( "OKX_OPTIONS_CLOSE_HOLD_SECONDS", "门控持续秒数", "条件达标后需持续满足的秒数,默认 120", ), ], } # 与运行时 os.getenv 默认一致;.env 未写明时展示实际生效值(同风控说明页) _RUNTIME_ENV_DEFAULTS: dict[str, str] = { "RISK_CONTROL_ENABLED": "true", "RISK_COOLING_HOURS_MANUAL": "4", "RISK_COOLING_HOURS_MANUAL_JOURNAL": "1", "RISK_MANUAL_CLOSE_DAILY_LIMIT": "2", "RISK_DAILY_LOSS_LIMIT": "2", "RISK_MOOD_ISSUES_DAILY_FREEZE": "true", "AUTO_TRANSFER_FROM": "funding", "AUTO_TRANSFER_TO": "swap", "TRANSFER_CCY": "USDT", "OKX_SHOW_PERP_FUNDS": "false", "OKX_OPTIONS_CHAIN_ASK_LIQ_FILTER_ENABLED": "true", "OKX_OPTIONS_CHAIN_MAX_DTE_DAYS": "14", "OKX_OPTIONS_MAX_DTE_DAYS": "2", "OKX_OPTIONS_MAX_ACTIVE_POSITIONS": "0", "OKX_OPTIONS_CLOSE_GATE_MODE": "premium", "OKX_OPTIONS_CLOSE_RECYCLE_MULT_COIN": "1.05", "OKX_OPTIONS_CLOSE_NET_PNL_MIN_USDT": "0", "OKX_OPTIONS_CLOSE_HOLD_SECONDS": "120", "OKX_TRADE_MODE": "options", } def _effective_env_value(key: str, file_values: dict[str, str], schema_default: str = "") -> str: if key == "OKX_TRADE_MODE": try: from lib.options.okx_trade_mode_lib import get_okx_trade_mode return get_okx_trade_mode() except Exception: return "options" if key == "OKX_OPTIONS_MARGIN_MODE": return "coin" if key in file_values: file_val = str(file_values.get(key) or "").strip() if file_val: return file_val runtime = os.getenv(key) if runtime is not None and str(runtime).strip() != "": return str(runtime).strip() if schema_default: return schema_default return _RUNTIME_ENV_DEFAULTS.get(key, "") def _env_truthy(raw: str) -> bool: return str(raw or "").strip().lower() in ("1", "true", "yes", "on") def _schema_field_map(example_path: str) -> dict[str, dict[str, Any]]: out: dict[str, dict[str, Any]] = {} for group in parse_env_example_schema(example_path): for field in group.get("fields") or []: out[field["key"]] = dict(field) return out def _build_field( key: str, label: str, note: str, schema: dict[str, dict[str, Any]], values: dict[str, str], ) -> dict[str, Any]: meta = schema.get(key) or {} schema_default = meta.get("default") or "" val = _effective_env_value(key, values, schema_default) masked = _mask_value(key, val) ftype = meta.get("type") or _field_type(key, val or schema_default) options = select_options_for(key) if options: ftype = "select" val = normalize_select_value(key, val) or val masked = _mask_value(key, val) out: dict[str, Any] = { "key": key, "label": label, "note": note or meta.get("note") or "", "default": val, "type": ftype, "sensitive": meta.get("sensitive", _is_sensitive(key)), "restart_required": meta.get("restart_required", _restart_required(key)), "hot_reload": meta.get("hot_reload", _hot_reload(key)), "current": masked["value"] if not _is_sensitive(key) else "", "masked": masked["masked"], "tail": masked.get("tail") or "", "has_value": masked["has_value"], } if options: cur = (out["current"] or out["default"] or "").strip() opt_vals = {o["value"] for o in options} if cur and cur not in opt_vals: options = [{"value": cur, "label": cur}] + options out["options"] = options return out def ui_sections_for_exchange( exchange_key: str, *, mode: str | None = None, ) -> list[dict[str, Any]]: del mode # 对冲模式已移除;保留参数兼容旧调用 ex = (exchange_key or "").strip().lower() sections: list[dict[str, Any]] = [] live_fields = _EXCHANGE_LIVE_FIELDS.get(ex, _EXCHANGE_LIVE_FIELDS["okx"]) sections.append({"title": "交易所与实盘", "fields": live_fields}) sections.extend(_SHARED_SECTIONS) if ex == "okx": sections.append({"title": "期权账户", "fields": list(_OPTIONS_SECTION["fields"])}) return sections def ui_allowed_keys(exchange_key: str) -> frozenset[str]: """可写键=当前可见字段。.env 里遗留对冲键仍保留在文件中,但 UI 不可再写入.""" keys: set[str] = set() for sec in ui_sections_for_exchange(exchange_key): for item in sec["fields"]: keys.add(item[0]) return frozenset(keys) def build_env_ui_payload( exchange_key: str, example_path: str, env_path: str, ) -> list[dict[str, Any]]: # 首次打开 env 配置时若 Token 为空则自动生成,便于复制到中控 try: from lib.hub.hub_bridge_auth import ensure_hub_bridge_token ensure_hub_bridge_token(env_path) except Exception: pass schema = _schema_field_map(example_path) env_lines = read_env_lines(env_path) values = env_get_all(env_lines) groups: list[dict[str, Any]] = [] for sec in ui_sections_for_exchange(exchange_key): fields = [ _build_field(key, label, note, schema, values) for key, label, note in sec["fields"] ] fields = _mark_options_section_hidden(fields) groups.append({ "title": sec["title"], "fields": fields, "has_restart": any(f.get("restart_required") for f in fields), }) return groups # 币本位专用:按复利/上限/门控模式标记 hidden def _field_current(fields: list[dict[str, Any]], key: str, default: str = "") -> str: for f in fields: if f.get("key") == key: return str(f.get("current") or f.get("default") or default).strip() return default def _mark_options_section_hidden(fields: list[dict[str, Any]]) -> list[dict[str, Any]]: """按门控模式/复利/单笔上限标记 hidden.""" gate_mode = (_field_current(fields, "OKX_OPTIONS_CLOSE_GATE_MODE", "premium") or "premium").lower() coin_compound_on = _env_truthy(_field_current(fields, "OKX_OPTIONS_COIN_COMPOUND", "true")) coin_max_on = _env_truthy(_field_current(fields, "OKX_OPTIONS_COIN_MAX_USDT_ENABLED", "false")) out: list[dict[str, Any]] = [] for f in fields: key = str(f.get("key") or "") hide = False if key == "OKX_OPTIONS_CLOSE_RECYCLE_MULT_COIN" and gate_mode != "premium": hide = True elif key == "OKX_OPTIONS_CLOSE_NET_PNL_MIN_USDT" and gate_mode != "net_pnl": hide = True elif key == "OKX_OPTIONS_COIN_BUDGET_USDT" and coin_compound_on: hide = True elif key == "OKX_OPTIONS_COIN_MAX_USDT" and not coin_max_on: hide = True if hide: item = dict(f) item["hidden"] = True out.append(item) else: out.append(f) return out def _mark_compound_budget_hidden(fields: list[dict[str, Any]]) -> list[dict[str, Any]]: """兼容旧调用名.""" return _mark_options_section_hidden(fields) def filter_updates_for_ui(exchange_key: str, updates: dict[str, str]) -> dict[str, str]: allowed = ui_allowed_keys(exchange_key) return {k: v for k, v in (updates or {}).items() if k in allowed} def validate_env_ui_updates( exchange_key: str, example_path: str, updates: dict[str, str], ) -> tuple[dict[str, str], list[str]]: from lib.env.env_schema import validate_env_updates schema = _schema_field_map(example_path) groups: list[dict[str, Any]] = [] for sec in ui_sections_for_exchange(exchange_key): fields: list[dict[str, Any]] = [] for key, _label, _note in sec["fields"]: if key in schema: field = dict(schema[key]) opts = select_options_for(key) if opts: field["type"] = "select" field["options"] = opts fields.append(field) else: default = "" fields.append( { "key": key, "type": _field_type(key, default), "sensitive": _is_sensitive(key), "restart_required": _restart_required(key), "hot_reload": _hot_reload(key), "options": select_options_for(key), } ) groups.append({"title": sec["title"], "fields": fields}) return validate_env_updates(groups, updates) def coerce_hedge_partial_close_with_manual( clean: dict[str, str], *, env_path: str = "", ) -> dict[str, str]: """兼容旧调用名;对冲已移除,原样返回.""" del env_path return dict(clean or {})