"""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, ) # 各所「交易所与实盘」字段(顺序即页面顺序) _OKX_LIVE_ONLY_KEYS = frozenset( { "LIVE_TRADING_ENABLED", "OKX_API_KEY", "OKX_API_SECRET", "OKX_API_PASSPHRASE", } ) _SIM_FUNDS_SECTION: dict[str, Any] = { "title": "模拟资金", "fields": [ ( "SIM_INITIAL_EQUITY_USDT", "初始权益 USDT", "重置模拟钱包时写入资金账户 USDT;改完需在系统设置「模拟资金」点重置才生效", ), ( "SIM_INITIAL_USDC", "初始 USDC", "重置时写入期权侧 USDC;改完需重置才生效", ), ( "SIM_FEE_RATE", "模拟手续费率", "如 0.0005=万五;撮合立即按此费率扣费", ), ], } _EXCHANGE_LIVE_FIELDS: dict[str, list[tuple[str, str, str]]] = { "okx": [ ( "SIM_DEFAULT_MODE", "撮合模式", "sim=本地模拟资金; live=实盘.保存后立即切换;sim 下隐藏 API/实盘开关", ), ("LIVE_TRADING_ENABLED", "开启实盘下单", "仅 live 模式下生效;关闭时即使 live 也不向交易所发单"), ("OKX_API_KEY", "API Key", "账户 API(永续+期权共用)"), ("OKX_API_SECRET", "API Secret", "账户 API(永续+期权共用)"), ("OKX_API_PASSPHRASE", "API Passphrase", "OKX 必填"), ("OKX_TD_MODE", "保证金模式", ""), ("OKX_POS_MODE", "持仓模式", ""), ("OKX_POSITION_INST_TYPE", "仓位查询类型", "如 SWAP"), ("OKX_ACCOUNT_LABEL", "账户备注", "企业微信推送中显示"), ( "OKX_SHOW_PERP_FUNDS", "显示永续资金", "默认开启;关闭后顶栏隐藏 USDT 资金账户与交易账户,总资金仅计期权 USDC 侧", ), ], "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", "账户备注", "企业微信推送中显示"), ], } _SHARED_SECTIONS: list[dict[str, Any]] = [ { "title": "企业微信", "fields": [ ("WECHAT_WEBHOOK", "机器人 Webhook", "行情与风控推送地址"), ("WECHAT_TIMEOUT_SECONDS", "推送超时(秒)", "默认 10"), ], }, ] _MODE_SECTION: dict[str, Any] = { "title": "期权/对冲模式", "exchanges": frozenset({"okx"}), "fields": [ ( "OKX_TRADE_MODE", "交易模式", "三选一:单独期权 / 永期对冲 / 期期对冲.选单独期权时隐藏对冲导航与对冲配置;选对冲时不可单独开期权", ), ], } _OPTIONS_SECTION: dict[str, Any] = { "title": "期权账户", "exchanges": frozenset({"okx"}), "fields": [ ("OKX_OPTIONS_ENABLED", "启用期权模块", "与永续共用上方 OKX_API_*;不再单独配置期权密钥"), ("OKX_OPTIONS_ACCOUNT_LABEL", "期权账户备注", ""), ( "OKX_OPTIONS_TRADE_BUDGET_USDC", "单笔预算(USDC)", "仅全仓复利关闭时显示/生效;用于「按可用余额打满」及张数/币数上限", ), ("OKX_OPTIONS_BUDGET_BUFFER", "预算缓冲比例", "如 0.95;打满/全仓复利共用"), ( "OKX_OPTIONS_COMPOUND_FULL_ENABLED", "全仓复利开关", "默认 true;开启时隐藏单笔预算且不可用打满预算,下单以全仓复利为主;关闭则恢复单笔预算并隐藏全仓复利", ), ( "OKX_OPTIONS_COMPOUND_FULL_CAP_ENABLED", "全仓复利上限开关", "仅全仓复利开启时有意义;默认 false=不设上限用期权户全部可用;true 时按下方上限封顶", ), ( "OKX_OPTIONS_COMPOUND_FULL_CAP_USDC", "全仓复利上限(USDC)", "仅「全仓复利」且「上限开关」都开启时生效;例如 300", ), ( "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_TRADE_MODE 取代的 ENABLED/SHOW/MUTUAL) _HEDGE_COMMON_FIELDS: list[tuple[str, str, str]] = [ ("HEDGE_PLAN_LIVE_ORDER", "允许对冲真实下单", "再与实盘 LIVE_TRADING_ENABLED 同开才可启动"), ( "MAX_ACTIVE_HEDGE_PLANS", "对冲组数上限", "默认 1;同时进行中的对冲计划组数(opening/active/partial),可改", ), ("HEDGE_PLAN_MONITOR_POLL_SECONDS", "对冲监控轮询(秒)", "默认 15"), ( "HEDGE_PLAN_BUDGET_BUFFER", "对冲预算缓冲比例", "默认 0.95;仅对冲计划;与期权页 OKX_OPTIONS_BUDGET_BUFFER 独立", ), ( "HEDGE_PLAN_MANUAL_COMPLETE_ON_PARTIAL", "半腿失败改手动补开", "默认 true;开启时半腿失败不自动平,计划挂 partial,页面可补开;并强制关闭下方自动平", ), ( "HEDGE_PLAN_PARTIAL_AUTO_CLOSE_OPTION", "半腿失败时自动平期权", "默认 true;若上方「半腿失败改手动补开」开启则本项强制无效", ), ] _HEDGE_PO_FIELDS: list[tuple[str, str, str]] = [ ( "HEDGE_PLAN_OPTION_PRIMARY", "永期模式(以期权为主/保险)", "默认 true=以期权为主;false=保险模式;页面标题前显示标识,不可在页内切换", ), ("POSITION_SIZING_MODE", "永续计仓模式", "切换须无仓后重启;以损定仓 / 全仓杠杆"), ("RISK_PERCENT", "以损定仓风险%", "永续腿单笔风险占资金比例"), ("FULL_MARGIN_BUFFER_RATIO", "全仓资金缓冲比例", "如 0.98"), ("BTC_LEVERAGE", "BTC 默认杠杆", "永续腿"), ("ALT_LEVERAGE", "山寨默认杠杆", "永续腿"), ("TRADING_DAY_RESET_HOUR", "交易日切点(北京时间)", "整点,默认 8;顶栏统计切日"), ("HEDGE_PLAN_OPEN_ORDER", "永期开仓顺序", "options_first 或 perp_first"), ("HEDGE_PLAN_ON_PERP_SL_CLOSE_OPTIONS", "永期止损后强制平期权", "保护机制,建议保持 true"), ("HEDGE_PLAN_ON_PERP_TP_CLOSE_OPTIONS", "永期止盈后强制平期权", "默认 false,保险腿不平"), ( "HEDGE_PLAN_ITM_MAX_DIST_USD", "永期实值最大深度(U)", "默认空=沿用 OKX_OPTIONS_ITM_MAX_DIST_USD(常 30);0=不限制", ), ( "HEDGE_PLAN_MIN_OPTION_HOURS", "对冲期权最低剩余小时", "默认 8;测算/启动时若传 hours_to_expiry 则校验", ), ( "HEDGE_PLAN_MIN_OPTION_LEVERAGE", "对冲期权最低杠杆(S/ask)", "默认 0=不启用;>0 时拒绝杠杆过低的保险腿", ), ] _HEDGE_OO_FIELDS: list[tuple[str, str, str]] = [ ("HEDGE_PLAN_OO_CLOSE_WINNER_ONLY", "期期只平盈利腿", "达目标价只平盈利方"), ( "HEDGE_PLAN_OO_CLOSE_MODE_ENABLED", "期期平仓模式(方案C)", "默认 true;开启后页面可选「到期平/全平」;关闭则固定到期平", ), ( "HEDGE_PLAN_OO_BIAS_SPLIT_BY", "期期做多做空拆分口径", "默认预算金额;budget=按权利金预算分两腿;sheets=先算同张数再按比例拆", ), ( "HEDGE_PLAN_OO_BIAS_RATIO", "期期做多做空主腿占比", "默认 0.7(即 7:3);做多主腿=Call,做空主腿=Put;须在 0~1 之间", ), ] # 兼容旧测试/全量字段列表(写 env 时仍允许这些键,但 UI 按模式过滤) _HEDGE_PLAN_SECTION: dict[str, Any] = { "title": "对冲计划", "exchanges": frozenset({"okx"}), "fields": [ ("HEDGE_PLAN_ENABLED", "启用对冲计划", "已由「交易模式」取代,一般无需再改"), ("HEDGE_PLAN_SHOW_PERP_OPTIONS", "显示永期对冲", "已由「交易模式」取代"), ("HEDGE_PLAN_SHOW_OPTIONS_OPTIONS", "显示期期对冲", "已由「交易模式」取代"), ("HEDGE_PLAN_OPTIONS_MUTUAL_EXCLUSIVE", "对冲与期权互斥门控", "已由「交易模式」三选一取代"), *_HEDGE_COMMON_FIELDS, *_HEDGE_PO_FIELDS, *_HEDGE_OO_FIELDS, ], } # 与运行时 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", "HEDGE_PLAN_SHOW_PERP_OPTIONS": "true", "HEDGE_PLAN_SHOW_OPTIONS_OPTIONS": "true", "OKX_SHOW_PERP_FUNDS": "true", "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_TRADE_MODE": "options", "SIM_DEFAULT_MODE": "sim", "MAX_ACTIVE_HEDGE_PLANS": "1", "HEDGE_PLAN_OO_CLOSE_MODE_ENABLED": "true", "HEDGE_PLAN_OO_BIAS_SPLIT_BY": "budget", "HEDGE_PLAN_OO_BIAS_RATIO": "0.7", "HEDGE_PLAN_BUDGET_BUFFER": "0.95", "HEDGE_PLAN_OPTION_PRIMARY": "true", "HEDGE_PLAN_OPTIONS_MUTUAL_EXCLUSIVE": "true", "HEDGE_PLAN_MANUAL_COMPLETE_ON_PARTIAL": "true", } def _effective_env_value(key: str, file_values: dict[str, str], schema_default: str = "") -> str: if key == "SIM_DEFAULT_MODE": # 展示当前生效撮合模式(运行时 trading.mode 优先) try: from lib.sim.mode_lib import peek_persisted_trading_mode cur = peek_persisted_trading_mode() if cur: return cur except Exception: pass if key == "OKX_TRADE_MODE": # 展示值必须与运行时 get_okx_trade_mode() 一致,避免未写入时默认 options 静默改模式 file_val = str(file_values.get(key) or "").strip() if key in file_values else "" if file_val: from lib.hedge_plan.okx_trade_mode_lib import normalize_okx_trade_mode return normalize_okx_trade_mode(file_val) or file_val try: from lib.hedge_plan.okx_trade_mode_lib import get_okx_trade_mode return get_okx_trade_mode() except Exception: pass 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) # 与运行时一致:手动补开开启时,「自动平期权」展示为关闭(实际也不会执行) if key == "HEDGE_PLAN_PARTIAL_AUTO_CLOSE_OPTION": manual = _effective_env_value( "HEDGE_PLAN_MANUAL_COMPLETE_ON_PARTIAL", values, "true" ) if _env_truthy(manual): val = "false" 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 _okx_mode_for_env_ui() -> str: try: from lib.hedge_plan.okx_trade_mode_lib import get_okx_trade_mode return get_okx_trade_mode() except Exception: return "options" def _options_fields_for_mode(mode: str) -> list[tuple[str, str, str]]: fields = list(_OPTIONS_SECTION["fields"]) if mode != "options": fields = [f for f in fields if f[0] != "OKX_OPTIONS_MAX_ACTIVE_POSITIONS"] return fields def _hedge_fields_for_mode(mode: str) -> list[tuple[str, str, str]]: if mode == "perp_options": return [*_HEDGE_COMMON_FIELDS, *_HEDGE_PO_FIELDS] if mode == "options_options": return [*_HEDGE_COMMON_FIELDS, *_HEDGE_OO_FIELDS] return [] def _trading_mode_for_env_ui() -> str: try: from lib.sim.mode_lib import peek_persisted_trading_mode, default_trading_mode return peek_persisted_trading_mode() or default_trading_mode() except Exception: return "sim" def _okx_exchange_fields_for_trading_mode(trading_mode: str) -> list[tuple[str, str, str]]: fields = list(_EXCHANGE_LIVE_FIELDS["okx"]) tm = (trading_mode or "").strip().lower() if tm == "sim": return [f for f in fields if f[0] not in _OKX_LIVE_ONLY_KEYS] return fields def ui_sections_for_exchange( exchange_key: str, *, mode: str | None = None, trading_mode: str | None = None, ) -> list[dict[str, Any]]: ex = (exchange_key or "").strip().lower() sections: list[dict[str, Any]] = [] tm = (trading_mode or "").strip().lower() if not tm: tm = _trading_mode_for_env_ui() if ex == "okx" else "live" if ex == "okx": live_fields = _okx_exchange_fields_for_trading_mode(tm) else: live_fields = _EXCHANGE_LIVE_FIELDS.get(ex, _EXCHANGE_LIVE_FIELDS["okx"]) sections.append({"title": "交易所与实盘", "fields": live_fields}) if ex == "okx" and tm == "sim": sections.append(_SIM_FUNDS_SECTION) sections.extend(_SHARED_SECTIONS) if ex in _MODE_SECTION.get("exchanges", frozenset()): from lib.hedge_plan.okx_trade_mode_lib import normalize_okx_trade_mode m = normalize_okx_trade_mode(mode) if mode else "" if not m: m = _okx_mode_for_env_ui() sections.append(_MODE_SECTION) sections.append({"title": "期权账户", "fields": _options_fields_for_mode(m)}) hedge_fields = _hedge_fields_for_mode(m) if hedge_fields: title = "对冲计划·永期" if m == "perp_options" else "对冲计划·期期" sections.append({"title": title, "fields": hedge_fields}) return sections def ui_allowed_keys(exchange_key: str) -> frozenset[str]: """可写键=当前模式可见字段 + 模式切换键 + 遗留对冲开关(兼容旧脚本写入).""" keys: set[str] = set() for sec in ui_sections_for_exchange(exchange_key): for item in sec["fields"]: keys.add(item[0]) ex = (exchange_key or "").strip().lower() if ex == "okx": keys.add("OKX_TRADE_MODE") keys.add("SIM_DEFAULT_MODE") # 切模式后同请求可能带上对侧字段,始终放行 keys.update(_OKX_LIVE_ONLY_KEYS) for item in _SIM_FUNDS_SECTION["fields"]: keys.add(item[0]) for item in _HEDGE_PLAN_SECTION["fields"]: keys.add(item[0]) for item in _OPTIONS_SECTION["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]]: schema = _schema_field_map(example_path) env_lines = read_env_lines(env_path) values = env_get_all(env_lines) trading_mode = "" if (exchange_key or "").strip().lower() == "okx": trading_mode = _effective_env_value("SIM_DEFAULT_MODE", values, "sim") or _trading_mode_for_env_ui() groups: list[dict[str, Any]] = [] for sec in ui_sections_for_exchange( exchange_key, mode=values.get("OKX_TRADE_MODE") or "", trading_mode=trading_mode, ): fields = [ _build_field(key, label, note, schema, values) for key, label, note in sec["fields"] ] fields = _mark_compound_budget_hidden(fields) groups.append({ "title": sec["title"], "fields": fields, "has_restart": any(f.get("restart_required") for f in fields), }) return groups def _mark_compound_budget_hidden(fields: list[dict[str, Any]]) -> list[dict[str, Any]]: """全仓复利开启时标记单笔预算为 hidden(供 SSR/前端隐藏;切换开关仍可再显示).""" compound_on = True for f in fields: if f.get("key") == "OKX_OPTIONS_COMPOUND_FULL_ENABLED": compound_on = _env_truthy(str(f.get("current") or f.get("default") or "true")) break if not compound_on: return fields out: list[dict[str, Any]] = [] for f in fields: if f.get("key") == "OKX_OPTIONS_TRADE_BUDGET_USDC": item = dict(f) item["hidden"] = True out.append(item) else: out.append(f) return out 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]: """手动补开为开启时,强制把自动平写成 false(与运行时一致).""" out = dict(clean or {}) manual = out.get("HEDGE_PLAN_MANUAL_COMPLETE_ON_PARTIAL") if manual is None and env_path: try: from lib.env.env_file_lib import env_get_all, read_env_lines file_vals = env_get_all(read_env_lines(env_path)) manual = _effective_env_value( "HEDGE_PLAN_MANUAL_COMPLETE_ON_PARTIAL", file_vals, "true" ) except Exception: manual = "true" if _env_truthy(str(manual or "")): out["HEDGE_PLAN_PARTIAL_AUTO_CLOSE_OPTION"] = "false" return out