"""env 配置页 UI 白名单：中文标签、按交易所过滤。""" from __future__ import annotations 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, parse_env_example_schema, ) # 各所「交易所与实盘」字段（顺序即页面顺序） _EXCHANGE_LIVE_FIELDS: dict[str, list[tuple[str, str, str]]] = { "okx": [ ("LIVE_TRADING_ENABLED", "开启实盘下单", "关闭时仅走本地流程，不向交易所发单"), ("OKX_API_KEY", "API Key", "永续子账户"), ("OKX_API_SECRET", "API Secret", "永续子账户"), ("OKX_API_PASSPHRASE", "API Passphrase", "OKX 必填"), ("OKX_TD_MODE", "保证金模式", "cross=全仓，isolated=逐仓"), ("OKX_POS_MODE", "持仓模式", "hedge=双向，net=单向净持仓"), ("OKX_POSITION_INST_TYPE", "仓位查询类型", "如 SWAP"), ("OKX_ACCOUNT_LABEL", "账户备注", "企业微信推送中显示"), ], "binance": [ ("LIVE_TRADING_ENABLED", "开启实盘下单", "关闭时仅走本地流程，不向交易所发单"), ("BINANCE_API_KEY", "API Key", "永续子账户"), ("BINANCE_API_SECRET", "API Secret", "永续子账户"), ("BINANCE_MARGIN_MODE", "保证金模式", "cross=全仓，isolated=逐仓"), ("BINANCE_POSITION_MODE", "持仓模式", "hedge=双向，one_way=单向"), ("BINANCE_ACCOUNT_LABEL", "账户备注", "企业微信推送中显示"), ], "gate": [ ("LIVE_TRADING_ENABLED", "开启实盘下单", "关闭时仅走本地流程，不向交易所发单"), ("GATE_API_KEY", "API Key", "永续子账户"), ("GATE_API_SECRET", "API Secret", "永续子账户"), ("GATE_TD_MODE", "保证金模式", "cross=全仓，isolated=逐仓"), ("GATE_POS_MODE", "持仓模式", "hedge=双向，single=单向"), ("GATE_ACCOUNT_LABEL", "账户备注", "企业微信推送中显示"), ], } _SHARED_SECTIONS: list[dict[str, Any]] = [ { "title": "企业微信", "fields": [ ("WECHAT_WEBHOOK", "机器人 Webhook", "行情与风控推送地址"), ("WECHAT_TIMEOUT_SECONDS", "推送超时（秒）", "默认 10"), ], }, { "title": "AI 复盘", "fields": [ ("AI_PROVIDER", "AI 提供方", "openai 或 ollama"), ("OPENAI_API_BASE", "API 地址", "OpenAI 兼容接口"), ("OPENAI_API_KEY", "API 密钥", "留空表示不修改"), ("OPENAI_MODEL", "云端模型", ""), ("OLLAMA_API", "Ollama 地址", "本地服务 URL"), ("AI_MODEL", "Ollama 模型", ""), ("AI_TIMEOUT_SECONDS", "请求超时（秒）", "默认 120"), ], }, { "title": "交易执行", "fields": [ ("POSITION_SIZING_MODE", "计仓模式", "risk=以损定仓，full_margin=全仓杠杆"), ("RISK_PERCENT", "以损定仓风险%", "单笔风险占资金比例"), ("FULL_MARGIN_BUFFER_RATIO", "全仓资金缓冲比例", "如 0.98"), ("BTC_LEVERAGE", "BTC 默认杠杆", ""), ("ALT_LEVERAGE", "山寨默认杠杆", ""), ("TRADE_DIRECTION_RESTRICT_ENABLED", "方向限制开关", ""), ("TRADE_DIRECTION", "允许方向", "long_only / short_only / both"), ("TRADE_SYMBOL_RESTRICT_ENABLED", "币种白名单开关", ""), ("TRADE_SYMBOL_WHITELIST", "白名单币种", "逗号分隔，如 BTC,ETH"), ("TRADING_DAY_RESET_HOUR", "交易日切点（北京时间）", "整点，默认 8"), ("TRADING_DAY_RESET_OPEN_GUARD_ENABLED", "切点前禁止新开仓", ""), ("MAX_ACTIVE_POSITIONS", "最大同时持仓", ""), ("MANUAL_MIN_PLANNED_RR", "人工最低盈亏比", "如 1.4"), ("FORCE_CLOSE_ENABLED", "强制清仓开关", ""), ("FORCE_CLOSE_BJ_HOUR", "强制清仓整点（北京）", ""), ], }, { "title": "交易风控", "fields": [ ("DAILY_OPEN_ALERT_THRESHOLD", "单日开仓提醒阈值", "达次数后 AI 提醒，不拦单"), ("DAILY_OPEN_HARD_LIMIT", "单日开仓硬上限", "0=不启用"), ], }, { "title": "账户冷静期", "fields": [ ("RISK_CONTROL_ENABLED", "冷静期总开关", ""), ("RISK_COOLING_HOURS_MANUAL", "手动平仓冷静（小时）", ""), ("RISK_COOLING_HOURS_MANUAL_JOURNAL", "复盘情绪冷静（小时）", ""), ("RISK_MANUAL_CLOSE_DAILY_LIMIT", "日手动平仓次数上限", ""), ("RISK_MOOD_ISSUES_DAILY_FREEZE", "情绪标签日冻结", ""), ], }, { "title": "自动划转", "fields": [ ("AUTO_TRANSFER_ENABLED", "启用自动划转", ""), ("AUTO_TRANSFER_AMOUNT", "目标余额（U）", "交易账户目标 USDT"), ("AUTO_TRANSFER_FROM", "划出账户", "funding 或 swap"), ("AUTO_TRANSFER_TO", "划入账户", "swap 或 funding"), ("AUTO_TRANSFER_BJ_HOUR", "执行整点（北京时间）", ""), ("TRANSFER_CCY", "划转币种", "默认 USDT"), ], }, { "title": "当日资金", "fields": [ ("DAILY_START_CAPITAL", "日起始基数（U）", ""), ("DAILY_LOSS_CAPITAL", "回撤后基数（U）", ""), ("DAILY_PROFIT_CAPITAL", "盈利后基数（U）", ""), ], }, ] _OPTIONS_SECTION: dict[str, Any] = { "title": "期权账户", "exchanges": frozenset({"okx"}), "fields": [ ("OKX_OPTIONS_ENABLED", "启用期权模块", ""), ("OKX_OPTIONS_API_KEY", "期权 API Key", "主账户，与永续子账户分离"), ("OKX_OPTIONS_API_SECRET", "期权 API Secret", ""), ("OKX_OPTIONS_API_PASSPHRASE", "期权 API Passphrase", ""), ("OKX_OPTIONS_ACCOUNT_LABEL", "期权账户备注", ""), ("OKX_OPTIONS_TRADE_BUDGET_USDC", "单笔预算（USDC）", ""), ("OKX_OPTIONS_BUDGET_BUFFER", "预算缓冲比例", "如 0.95"), ("OKX_OPTIONS_DEFAULT_UNDERLY", "默认标的", "如 ETH"), ], } 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 {} default_val = meta.get("default") or values.get(key) or "" val = values.get(key) if val is None: val = default_val masked = _mask_value(key, val) ftype = meta.get("type") or _field_type(key, default_val) return { "key": key, "label": label, "note": note or meta.get("note") or "", "default": 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"], } def ui_sections_for_exchange(exchange_key: str) -> list[dict[str, Any]]: 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 in _OPTIONS_SECTION.get("exchanges", frozenset()): sections.append(_OPTIONS_SECTION) 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]) 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) 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"] ] groups.append({ "title": sec["title"], "fields": fields, "has_restart": any(f.get("restart_required") for f in fields), }) return groups 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: fields.append(schema[key]) 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), } ) groups.append({"title": sec["title"], "fields": fields}) return validate_env_updates(groups, updates)