crypto_okx/lib/instance/instance_settings_lib.py

"""实例「系统设置」页:从 .env 汇总风控说明(三所共用)."""
from __future__ import annotations

import os
from typing import Any, Optional

from lib.trade.trade_policy_lib import TradePolicy


def _env_bool(key: str, default: bool = False) -> bool:
    raw = (os.getenv(key) or "").strip().lower()
    if not raw:
        return default
    return raw in ("1", "true", "yes", "on")


def _env_float(key: str, default: float) -> float:
    try:
        return float(os.getenv(key, str(default)))
    except (TypeError, ValueError):
        return default


def _env_int(key: str, default: int) -> int:
    try:
        return int(os.getenv(key, str(default)))
    except (TypeError, ValueError):
        return default


def _row(label: str, value: str, note: str = "") -> dict[str, str]:
    return {"label": label, "value": value, "note": note}


def _on_off(enabled: bool) -> str:
    return "开启" if enabled else "关闭"


def build_instance_settings_view(
    *,
    exchange_key: str,
    exchange_display: str,
    risk_status: Optional[dict[str, Any]] = None,
    trade_policy: Optional[TradePolicy] = None,
    data_export_version: int = 3,
    open_guard_enabled: Optional[bool] = None,
) -> dict[str, Any]:
    rs = risk_status or {}
    force_close_on = _env_bool("FORCE_CLOSE_ENABLED", False)
    force_close_hour = _env_int("FORCE_CLOSE_BJ_HOUR", 0)
    auto_transfer_on = _env_bool("AUTO_TRANSFER_ENABLED", False)

    sections: list[dict[str, Any]] = []

    # 期权/对冲独立版:不再展示永续「交易执行」「账户冷静期」说明

    sections.append(
        {
            "title": "关键位监控",
            "rows": [
                _row("模式", "仅关键支撑阻力提醒", "箱体/斐波/触价等程序自动单已移除"),
            ],
        }
    )

    if force_close_on or (exchange_key or "").strip().lower() == "gate":
        sections.append(
            {
                "title": "整点强制清仓",
                "rows": [
                    _row("强制清仓", _on_off(force_close_on)),
                    _row(
                        "执行时刻",
                        f"北京时间 {force_close_hour}:00 起 {_env_int('FORCE_CLOSE_GRACE_MINUTES', 5)} 分钟内",
                    ),
                ],
            }
        )

    if (exchange_key or "").strip().lower() == "okx" and _env_bool("OKX_OPTIONS_ENABLED", True):
        api_key = (os.getenv("OKX_API_KEY") or "").strip()
        sections.append(
            {
                "title": "期权设置",
                "rows": [
                    _row("期权模块", "已启用"),
                    _row(
                        "账户 API",
                        f"已配置(…{api_key[-4:]})" if len(api_key) >= 4 else "未配置 OKX_API_*",
                        "永续与期权共用 OKX_API_*",
                    ),
                    _row(
                        "说明",
                        "币种兑换与账户内划转到右侧「期权设置」卡片操作",
                    ),
                ],
            }
        )

    policy_note = ""
    if trade_policy and getattr(trade_policy, "badge_text", ""):
        policy_note = str(trade_policy.badge_text)

    return {
        "exchange_display": exchange_display,
        "risk_status_label": str(rs.get("status_label") or "正常"),
        "risk_status_reason": str(rs.get("reason") or "").strip(),
        "can_trade": bool(rs.get("can_trade", True)),
        "trade_policy_note": policy_note,
        "sections": sections,
        "data_export_version": int(data_export_version),
        "show_transfer": (exchange_key or "").strip().lower() in ("gate", "binance", "okx"),
        "options_settings_enabled": (exchange_key or "").strip().lower() == "okx"
        and _env_bool("OKX_OPTIONS_ENABLED", True),
        "auto_transfer_enabled": auto_transfer_on,
        "auto_transfer_bj_hour": _env_int("AUTO_TRANSFER_BJ_HOUR", 8),
        "auto_transfer_amount": _env_float("AUTO_TRANSFER_AMOUNT", 30),
        "auto_transfer_from": (os.getenv("AUTO_TRANSFER_FROM") or "funding").strip(),
        "auto_transfer_to": (os.getenv("AUTO_TRANSFER_TO") or "swap").strip(),
    }


def build_settings_tabs(display: dict[str, Any] | None, instance_settings: dict[str, Any]) -> list[dict[str, str]]:
    disp = display or {}
    inst = instance_settings or {}
    tabs: list[dict[str, str]] = [{"key": "nav", "title": "导航显示"}]
    tabs.append({"key": "sim_funds", "title": "模拟资金"})
    if disp.get("show_settings_password", True):
        tabs.append({"key": "password", "title": "账户密码"})
    if inst.get("show_transfer") and disp.get("show_settings_transfer", True):
        tabs.append({"key": "transfer", "title": "永续划转"})
    if disp.get("show_settings_export", True):
        tabs.append({"key": "export", "title": "数据导出"})
    if inst.get("options_settings_enabled") and disp.get("show_settings_options_swap", True):
        tabs.append({"key": "options_swap", "title": "币种兑换"})
    if inst.get("options_settings_enabled") and disp.get("show_settings_options_transfer", True):
        tabs.append({"key": "options_transfer", "title": "期权划转"})
    return tabs


def settings_page_context(page: str, *, instance_base_dir: str | None = None, **kwargs: Any) -> dict[str, Any]:
    p = (page or "").strip()
    if p == "system_guide":
        from lib.instance.instance_system_guide_lib import system_guide_template_context

        return system_guide_template_context()
    if p not in ("settings", "risk_policy", "env_config"):
        return {}
    display = kwargs.pop("display", None)
    ctx: dict[str, Any] = {"instance_settings": build_instance_settings_view(**kwargs)}
    if p == "settings":
        ctx["settings_tabs"] = build_settings_tabs(display, ctx["instance_settings"])
    if p == "env_config" and instance_base_dir:
        from lib.env.env_ui_manifest import build_env_ui_payload

        exchange_key = str(kwargs.get("exchange_key") or "")
        env_path = os.path.join(instance_base_dir, ".env")
        example_path = os.path.join(instance_base_dir, ".env.example")
        ctx["env_config_groups"] = build_env_ui_payload(exchange_key, example_path, env_path)
    return ctx