"""实例「系统设置」页:从 .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 分钟内"), ], } ) if (exchange_key or "").strip().lower() == "okx" and _env_bool("OKX_OPTIONS_ENABLED", False): opt_key = (os.getenv("OKX_OPTIONS_API_KEY") or "").strip() sections.append( { "title": "期权账户(主账户)", "rows": [ _row("期权模块", "已启用"), _row( "期权 API", f"已配置(…{opt_key[-4:]})" if len(opt_key) >= 4 else "未配置", ), _row("单笔权利金上限", f"{_env_float('OKX_OPTIONS_TRADE_BUDGET_USDC', 10):g} USDC"), _row( "资金说明", "资金账户 USDT 兑换 USDC 后划转到交易账户", "期权页操作;与永续子账户资金分开", ), ], } ) 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)}