crypto_monitor/lib/instance/instance_settings_lib.py

"""实例「系统设置」页：从 .env 汇总风控说明（三所共用）。"""
from __future__ import annotations

import os
from typing import Any, Optional

from lib.key_monitor.key_auto_order_lib import load_key_auto_order_enabled
from lib.trade.account_risk_lib import (
    cooling_hours_manual,
    cooling_hours_manual_journal,
    manual_close_daily_limit,
    max_active_positions_from_env,
    mood_issues_daily_freeze_enabled,
    risk_control_enabled,
)
from lib.trade.position_sizing_lib import is_full_margin_mode, load_position_sizing_mode, mode_label_zh
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,
) -> dict[str, Any]:
    rs = risk_status or {}
    sizing_mode = load_position_sizing_mode()
    key_auto = load_key_auto_order_enabled()
    reset_hour = _env_int("TRADING_DAY_RESET_HOUR", 8)
    hard_limit = _env_int("DAILY_OPEN_HARD_LIMIT", 0)
    alert_threshold = _env_int("DAILY_OPEN_ALERT_THRESHOLD", 5)
    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("最大同时持仓", str(max_active_positions_from_env())),
                _row("计仓模式", mode_label_zh(sizing_mode)),
                _row("以损定仓风险%", f"{_env_float('RISK_PERCENT', 2):g}%"),
                _row("人工最低盈亏比", f">= {_env_float('MANUAL_MIN_PLANNED_RR', 1.4):g}:1"),
                _row(
                    "交易日切点",
                    f"北京时间 {reset_hour}:00",
                    "新交易日统计与部分开仓限制以此为准",
                ),
                _row(
                    "单日开仓提醒",
                    f"第 {alert_threshold} 次",
                    "达次数推送企业微信，不拦单",
                ),
                _row(
                    "单日开仓硬上限",
                    str(hard_limit) if hard_limit > 0 else "未启用",
                    "达上限后禁止一切新开仓直至下一交易日" if hard_limit > 0 else "",
                ),
            ],
        }
    )

    sections.append(
        {
            "title": "账户冷静期",
            "rows": [
                _row("风控总开关", _on_off(risk_control_enabled())),
                _row("手动平仓冷静", f"{cooling_hours_manual():g} 小时"),
                _row("复盘后冷静", f"{cooling_hours_manual_journal():g} 小时", "手动平仓且填写说明后可缩短"),
                _row("日手动平仓上限", f"{manual_close_daily_limit()} 次", "超限当日冻结"),
                _row(
                    "复盘情绪日冻结",
                    _on_off(mood_issues_daily_freeze_enabled()),
                    "复盘勾选心态标签可触发当日冻结",
                ),
            ],
        }
    )

    key_rows = [
        _row("关键位自动单", _on_off(key_auto)),
        _row("关键位最低盈亏比", f"> {_env_float('KEY_AUTO_MIN_PLANNED_RR', 1.5):g}:1"),
    ]
    if is_full_margin_mode(sizing_mode):
        key_rows.append(
            _row(
                "全仓模式",
                "仅触价类自动单",
                "箱体/斐波等自动开仓在全仓下禁用",
            )
        )
    sections.append({"title": "关键位与自动单", "rows": key_rows})

    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 起 15 分钟内"),
                ],
            }
        )

    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"),
        "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 settings_page_context(page: str, **kwargs: Any) -> dict[str, Any]:
    if (page or "").strip() != "settings":
        return {}
    return {"instance_settings": build_instance_settings_view(**kwargs)}