dekun
435 lines
14 KiB
Python
435 lines
14 KiB
Python
"""从 .env.example 构建 env 配置 schema(分组,敏感,重启标注)."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
from typing import Any, Optional
|
|
|
|
from lib.env.env_file_lib import env_get, env_get_all, read_env_lines
|
|
|
|
_GROUP_RE = re.compile(r"^#\s*=+\s*(.+?)\s*=+\s*$")
|
|
_SEPARATOR_RE = re.compile(r"^#\s*=+\s*$")
|
|
_SECTION_DASH_RE = re.compile(r"^#\s*---\s*(.+?)\s*---\s*$")
|
|
_KEY_LINE = re.compile(r"^([A-Za-z_][A-Za-z0-9_]*)\s*=")
|
|
|
|
RESTART_REQUIRED_EXACT = frozenset({
|
|
"APP_HOST",
|
|
"APP_PORT",
|
|
"APP_DEBUG",
|
|
"DB_PATH",
|
|
"UPLOAD_DIR",
|
|
"FLASK_SECRET_KEY",
|
|
"POSITION_SIZING_MODE",
|
|
"LIVE_TRADING_ENABLED",
|
|
"OKX_TD_MODE",
|
|
"OKX_POS_MODE",
|
|
"OKX_POSITION_INST_TYPE",
|
|
"BINANCE_MARGIN_MODE",
|
|
"BINANCE_POSITION_MODE",
|
|
"GATE_TD_MODE",
|
|
"GATE_POS_MODE",
|
|
"PM2_APP_NAME",
|
|
})
|
|
|
|
RESTART_REQUIRED_PREFIXES = (
|
|
"OKX_API_",
|
|
"OKX_OPTIONS_API_",
|
|
"BINANCE_API_",
|
|
"GATE_API_",
|
|
"OKX_SOCKS_",
|
|
"OKX_HTTP_",
|
|
"OKX_HTTPS_",
|
|
"BINANCE_HTTP_",
|
|
"BINANCE_HTTPS_",
|
|
"GATE_HTTP_",
|
|
"GATE_HTTPS_",
|
|
)
|
|
|
|
HOT_RELOAD_EXACT = frozenset({
|
|
"RISK_PERCENT",
|
|
"MAX_ACTIVE_POSITIONS",
|
|
"MANUAL_MIN_PLANNED_RR",
|
|
"KEY_AUTO_MIN_PLANNED_RR",
|
|
"DAILY_OPEN_ALERT_THRESHOLD",
|
|
"DAILY_OPEN_HARD_LIMIT",
|
|
"TRADING_DAY_RESET_HOUR",
|
|
"TRADING_DAY_RESET_OPEN_GUARD_ENABLED",
|
|
"RISK_CONTROL_ENABLED",
|
|
"RISK_COOLING_HOURS_MANUAL",
|
|
"RISK_COOLING_HOURS_MANUAL_JOURNAL",
|
|
"RISK_MANUAL_CLOSE_DAILY_LIMIT",
|
|
"RISK_DAILY_LOSS_LIMIT",
|
|
"RISK_MOOD_ISSUES_DAILY_FREEZE",
|
|
"KEY_AUTO_ORDER_ENABLED",
|
|
"TRADE_DIRECTION_RESTRICT_ENABLED",
|
|
"TRADE_DIRECTION",
|
|
"TRADE_SYMBOL_RESTRICT_ENABLED",
|
|
"TRADE_SYMBOL_WHITELIST",
|
|
"BALANCE_REFRESH_SECONDS",
|
|
"PRICE_REFRESH_SECONDS",
|
|
"MONITOR_POLL_SECONDS",
|
|
"AUTO_TRANSFER_ENABLED",
|
|
"AUTO_TRANSFER_AMOUNT",
|
|
"AUTO_TRANSFER_FROM",
|
|
"AUTO_TRANSFER_TO",
|
|
"AUTO_TRANSFER_BJ_HOUR",
|
|
"TRANSFER_CCY",
|
|
"FORCE_CLOSE_ENABLED",
|
|
"FORCE_CLOSE_BJ_HOUR",
|
|
"FORCE_CLOSE_GRACE_MINUTES",
|
|
"BTC_LEVERAGE",
|
|
"ALT_LEVERAGE",
|
|
"DAILY_START_CAPITAL",
|
|
"DAILY_LOSS_CAPITAL",
|
|
"DAILY_PROFIT_CAPITAL",
|
|
"FULL_MARGIN_BUFFER_RATIO",
|
|
"APP_USERNAME",
|
|
"APP_PASSWORD",
|
|
"APP_AUTH_DISABLED",
|
|
"WECHAT_WEBHOOK",
|
|
"HEDGE_PLAN_ENABLED",
|
|
"HEDGE_PLAN_SHOW_PERP_OPTIONS",
|
|
"HEDGE_PLAN_SHOW_OPTIONS_OPTIONS",
|
|
"OKX_SHOW_PERP_FUNDS",
|
|
"OKX_OPTIONS_CHAIN_ASK_LIQ_FILTER_ENABLED",
|
|
"OKX_OPTIONS_CHAIN_MAX_DTE_DAYS",
|
|
"OKX_OPTIONS_MAX_DTE_DAYS",
|
|
"OKX_OPTIONS_MAX_ACTIVE_POSITIONS",
|
|
"OKX_OPTIONS_COMPOUND_FULL_ENABLED",
|
|
"OKX_OPTIONS_COMPOUND_FULL_CAP_ENABLED",
|
|
"OKX_OPTIONS_COMPOUND_FULL_CAP_USDC",
|
|
"OKX_OPTIONS_TRADE_BUDGET_USDC",
|
|
"OKX_OPTIONS_BUDGET_BUFFER",
|
|
"OKX_OPTIONS_COIN_COMPOUND",
|
|
"OKX_OPTIONS_COIN_BUDGET_USDT",
|
|
"OKX_OPTIONS_COIN_MAX_USDT_ENABLED",
|
|
"OKX_OPTIONS_COIN_MAX_USDT",
|
|
"OKX_OPTIONS_COIN_SPOT_BUY_BUFFER",
|
|
"OKX_TRADE_MODE",
|
|
"MAX_ACTIVE_HEDGE_PLANS",
|
|
"HEDGE_PLAN_LIVE_ORDER",
|
|
"HEDGE_PLAN_OPTION_PRIMARY",
|
|
"HEDGE_PLAN_OPEN_ORDER",
|
|
"HEDGE_PLAN_ON_PERP_SL_CLOSE_OPTIONS",
|
|
"HEDGE_PLAN_ON_PERP_TP_CLOSE_OPTIONS",
|
|
"HEDGE_PLAN_OO_CLOSE_WINNER_ONLY",
|
|
"HEDGE_PLAN_OO_CLOSE_MODE_ENABLED",
|
|
"HEDGE_PLAN_OO_BIAS_SPLIT_BY",
|
|
"HEDGE_PLAN_OO_BIAS_RATIO",
|
|
"HEDGE_PLAN_BUDGET_BUFFER",
|
|
"HEDGE_PLAN_OPTIONS_MUTUAL_EXCLUSIVE",
|
|
"HEDGE_PLAN_MANUAL_COMPLETE_ON_PARTIAL",
|
|
"MAX_ACTIVE_HEDGE_PLANS",
|
|
"HEDGE_PLAN_MONITOR_POLL_SECONDS",
|
|
"HEDGE_PLAN_PARTIAL_AUTO_CLOSE_OPTION",
|
|
})
|
|
|
|
SENSITIVE_EXACT = frozenset({
|
|
"APP_PASSWORD",
|
|
"FLASK_SECRET_KEY",
|
|
"HUB_BRIDGE_TOKEN",
|
|
"OPENAI_API_KEY",
|
|
})
|
|
|
|
SENSITIVE_SUBSTR = ("_SECRET", "_PASSPHRASE", "_API_KEY", "_PASSWORD")
|
|
|
|
# env 配置页下拉:value → 中文标签
|
|
SELECT_OPTIONS: dict[str, tuple[tuple[str, str], ...]] = {
|
|
"OKX_TD_MODE": (("cross", "全仓"), ("isolated", "逐仓")),
|
|
"OKX_POS_MODE": (("hedge", "双向"), ("net", "单向净持仓")),
|
|
"BINANCE_MARGIN_MODE": (("cross", "全仓"), ("isolated", "逐仓")),
|
|
"BINANCE_POSITION_MODE": (("hedge", "双向"), ("one_way", "单向")),
|
|
"GATE_TD_MODE": (("cross", "全仓"), ("isolated", "逐仓")),
|
|
"GATE_POS_MODE": (("hedge", "双向"), ("single", "单向")),
|
|
"POSITION_SIZING_MODE": (("risk", "以损定仓"), ("full_margin", "全仓杠杆")),
|
|
"TRADE_DIRECTION": (
|
|
("both", "双向均可"),
|
|
("long_only", "仅做多"),
|
|
("short_only", "仅做空"),
|
|
),
|
|
"AUTO_TRANSFER_FROM": (
|
|
("funding", "funding 资金账户"),
|
|
("swap", "swap 交易账户"),
|
|
("spot", "spot 现货"),
|
|
),
|
|
"AUTO_TRANSFER_TO": (
|
|
("swap", "swap 交易账户"),
|
|
("funding", "funding 资金账户"),
|
|
("spot", "spot 现货"),
|
|
),
|
|
"TRANSFER_CCY": (("USDT", "USDT"),),
|
|
"HEDGE_PLAN_OO_BIAS_SPLIT_BY": (
|
|
("budget", "预算金额"),
|
|
("sheets", "张数"),
|
|
),
|
|
"OKX_TRADE_MODE": (
|
|
("options", "单独期权"),
|
|
("perp_options", "永期对冲"),
|
|
("options_options", "期期对冲"),
|
|
),
|
|
"OKX_OPTIONS_MARGIN_MODE": (
|
|
("coin", "币本位(USDT买币桥)"),
|
|
("usdc", "USDC(USDⓈ权利金)"),
|
|
),
|
|
"OKX_OPTIONS_CLOSE_GATE_MODE": (
|
|
("premium", "权利金×倍数"),
|
|
("net_pnl", "净盈亏(U)阈值"),
|
|
),
|
|
"HEDGE_PLAN_OPTION_PRIMARY": (
|
|
("true", "以期权为主"),
|
|
("false", "保险模式"),
|
|
),
|
|
}
|
|
|
|
_SELECT_ALIASES: dict[str, dict[str, str]] = {
|
|
"OKX_TD_MODE": {"cross_margin": "cross", "isolated_margin": "isolated"},
|
|
"BINANCE_MARGIN_MODE": {"cross_margin": "cross", "isolated_margin": "isolated"},
|
|
"GATE_TD_MODE": {"cross_margin": "cross", "isolated_margin": "isolated"},
|
|
"TRANSFER_CCY": {"usdt": "USDT"},
|
|
}
|
|
|
|
|
|
def _is_sensitive(key: str) -> bool:
|
|
if key in SENSITIVE_EXACT:
|
|
return True
|
|
return any(s in key for s in SENSITIVE_SUBSTR)
|
|
|
|
|
|
def select_options_for(key: str) -> list[dict[str, str]]:
|
|
opts = SELECT_OPTIONS.get(key) or ()
|
|
return [{"value": v, "label": lab} for v, lab in opts]
|
|
|
|
|
|
def normalize_select_value(key: str, value: Optional[str]) -> str:
|
|
raw = (value or "").strip()
|
|
if not raw:
|
|
return ""
|
|
low = raw.lower()
|
|
aliases = _SELECT_ALIASES.get(key) or {}
|
|
if low in aliases:
|
|
return aliases[low]
|
|
allowed = {v for v, _ in (SELECT_OPTIONS.get(key) or ())}
|
|
allowed_by_lower = {v.lower(): v for v in allowed}
|
|
if low in allowed:
|
|
return low
|
|
if raw in allowed:
|
|
return raw
|
|
if low in allowed_by_lower:
|
|
return allowed_by_lower[low]
|
|
return raw
|
|
|
|
|
|
def _restart_required(key: str) -> bool:
|
|
if key in HOT_RELOAD_EXACT:
|
|
return False
|
|
if key in RESTART_REQUIRED_EXACT:
|
|
return True
|
|
return any(key.startswith(p) for p in RESTART_REQUIRED_PREFIXES)
|
|
|
|
|
|
def _hot_reload(key: str) -> bool:
|
|
if key in HOT_RELOAD_EXACT:
|
|
return True
|
|
if _restart_required(key):
|
|
return False
|
|
return key.startswith(("KEY_", "KLINE_", "BREAKEVEN_", "RECONCILE_", "ORDER_CHART_"))
|
|
|
|
|
|
def _field_type(key: str, value: str) -> str:
|
|
if key in SELECT_OPTIONS:
|
|
return "select"
|
|
low = (value or "").strip().lower()
|
|
if low in ("true", "false"):
|
|
return "bool"
|
|
if key.endswith("_ENABLED") or key.startswith("RISK_MOOD_") or key in (
|
|
"OKX_SHOW_PERP_FUNDS",
|
|
"HEDGE_PLAN_SHOW_PERP_OPTIONS",
|
|
"HEDGE_PLAN_SHOW_OPTIONS_OPTIONS",
|
|
):
|
|
return "bool"
|
|
try:
|
|
if "." in low:
|
|
float(low)
|
|
return "float"
|
|
int(low)
|
|
return "int"
|
|
except ValueError:
|
|
pass
|
|
return "text"
|
|
|
|
|
|
def _mask_value(key: str, value: Optional[str]) -> dict[str, Any]:
|
|
if value is None or value == "":
|
|
return {"value": "", "masked": "", "tail": "", "has_value": False}
|
|
if not _is_sensitive(key):
|
|
return {"value": value, "masked": value, "tail": "", "has_value": True}
|
|
tail = value[-4:] if len(value) >= 4 else value
|
|
return {"value": "", "masked": f"****{tail}", "tail": tail, "has_value": True}
|
|
|
|
|
|
def parse_env_example_schema(example_path: str) -> list[dict[str, Any]]:
|
|
if not os.path.isfile(example_path):
|
|
return []
|
|
lines = read_env_lines(example_path)
|
|
groups: list[dict[str, Any]] = []
|
|
group_map: dict[str, dict[str, Any]] = {}
|
|
current_group = "基础配置"
|
|
pending_note: list[str] = []
|
|
in_section_block = False
|
|
section_title_set = False
|
|
allow_section_blocks = False
|
|
|
|
def _ensure_group(title: str) -> dict[str, Any]:
|
|
title = (title or "").strip() or "其他"
|
|
if title not in group_map:
|
|
group_map[title] = {"title": title, "fields": []}
|
|
groups.append(group_map[title])
|
|
return group_map[title]
|
|
|
|
for raw in lines:
|
|
line = raw.rstrip()
|
|
stripped = line.strip()
|
|
if not stripped:
|
|
pending_note = []
|
|
continue
|
|
if _SEPARATOR_RE.match(stripped):
|
|
if not allow_section_blocks:
|
|
continue
|
|
if not in_section_block:
|
|
in_section_block = True
|
|
section_title_set = False
|
|
else:
|
|
in_section_block = False
|
|
continue
|
|
if in_section_block and stripped.startswith("#"):
|
|
note = stripped.lstrip("#").strip()
|
|
if note and not section_title_set:
|
|
current_group = note
|
|
_ensure_group(current_group)
|
|
section_title_set = True
|
|
elif note:
|
|
pending_note.append(note)
|
|
continue
|
|
gm = _GROUP_RE.match(stripped)
|
|
if gm:
|
|
title = gm.group(1).strip()
|
|
if title and title != "=":
|
|
current_group = title
|
|
_ensure_group(current_group)
|
|
in_section_block = False
|
|
section_title_set = False
|
|
pending_note = []
|
|
continue
|
|
dash = _SECTION_DASH_RE.match(stripped)
|
|
if dash:
|
|
allow_section_blocks = True
|
|
current_group = dash.group(1).strip()
|
|
_ensure_group(current_group)
|
|
in_section_block = False
|
|
section_title_set = False
|
|
pending_note = []
|
|
continue
|
|
if stripped.startswith("#"):
|
|
note = stripped.lstrip("#").strip()
|
|
if note and not note.startswith("="):
|
|
pending_note.append(note)
|
|
continue
|
|
km = _KEY_LINE.match(stripped)
|
|
if not km:
|
|
continue
|
|
key = km.group(1)
|
|
allow_section_blocks = True
|
|
default_val = env_get(lines, key) or ""
|
|
grp = _ensure_group(current_group)
|
|
note = " ".join(pending_note).strip()
|
|
grp["fields"].append(
|
|
{
|
|
"key": key,
|
|
"label": key,
|
|
"note": note,
|
|
"default": default_val,
|
|
"type": _field_type(key, default_val),
|
|
"sensitive": _is_sensitive(key),
|
|
"restart_required": _restart_required(key),
|
|
"hot_reload": _hot_reload(key),
|
|
}
|
|
)
|
|
pending_note = []
|
|
return [g for g in groups if g.get("fields")]
|
|
|
|
|
|
def build_env_payload(example_path: str, env_path: str) -> dict[str, Any]:
|
|
groups = parse_env_example_schema(example_path)
|
|
env_lines = read_env_lines(env_path)
|
|
values = env_get_all(env_lines)
|
|
for group in groups:
|
|
for field in group.get("fields") or []:
|
|
key = field["key"]
|
|
val = values.get(key)
|
|
if val is None:
|
|
val = field.get("default") or ""
|
|
masked = _mask_value(key, val)
|
|
field["current"] = masked["value"] if not field["sensitive"] else ""
|
|
field["masked"] = masked["masked"]
|
|
field["has_value"] = masked["has_value"]
|
|
return {"groups": groups}
|
|
|
|
|
|
def validate_env_updates(groups: list[dict], updates: dict[str, str]) -> tuple[dict[str, str], list[str]]:
|
|
allowed = {}
|
|
for group in groups:
|
|
for field in group.get("fields") or []:
|
|
allowed[field["key"]] = field
|
|
clean: dict[str, str] = {}
|
|
errors: list[str] = []
|
|
for key, value in (updates or {}).items():
|
|
if key not in allowed:
|
|
errors.append(f"未知配置项: {key}")
|
|
continue
|
|
if value is None:
|
|
continue
|
|
val = str(value).strip()
|
|
if allowed[key].get("sensitive") and (val == "" or (val.startswith("****") and len(val) <= 8)):
|
|
continue
|
|
# API Key 被密码管理器/自动填充成登录密码时通常很短;OKX Key 一般为 36 位
|
|
if key.endswith("_API_KEY") and 0 < len(val) < 16:
|
|
errors.append(f"{key} 长度异常,疑似自动填充;留空则不修改已有密钥")
|
|
continue
|
|
ftype = allowed[key].get("type")
|
|
if ftype == "bool":
|
|
low = val.lower()
|
|
if low not in ("true", "false", "1", "0", "yes", "no", "on", "off"):
|
|
errors.append(f"{key} 须为 true/false")
|
|
continue
|
|
val = "true" if low in ("true", "1", "yes", "on") else "false"
|
|
elif ftype == "select" or key in SELECT_OPTIONS:
|
|
allowed_vals = {
|
|
str(o.get("value") if isinstance(o, dict) else o[0]).lower()
|
|
for o in (allowed[key].get("options") or select_options_for(key))
|
|
}
|
|
norm = normalize_select_value(key, val)
|
|
if allowed_vals and norm.lower() not in allowed_vals:
|
|
labels = " / ".join(
|
|
f"{o['value']}({o['label']})" if isinstance(o, dict) else f"{o[0]}({o[1]})"
|
|
for o in (allowed[key].get("options") or select_options_for(key))
|
|
)
|
|
errors.append(f"{key} 须为: {labels}")
|
|
continue
|
|
val = norm
|
|
clean[key] = val
|
|
return clean, errors
|
|
|
|
|
|
def updates_need_restart(groups: list[dict], changed_keys: list[str]) -> bool:
|
|
field_map = {}
|
|
for group in groups:
|
|
for field in group.get("fields") or []:
|
|
field_map[field["key"]] = field
|
|
for key in changed_keys:
|
|
meta = field_map.get(key) or {}
|
|
if meta.get("restart_required"):
|
|
return True
|
|
if not meta.get("hot_reload"):
|
|
return True
|
|
return False
|