""" 四所共用:计仓模式 risk(以损定仓)| full_margin(全仓杠杆)。 仅 env POSITION_SIZING_MODE 切换;须无持仓(由部署流程保证)。 """ from __future__ import annotations import os from typing import Any, Optional, Tuple MODE_RISK = "risk" MODE_FULL_MARGIN = "full_margin" VALID_MODES = frozenset({MODE_RISK, MODE_FULL_MARGIN}) OPEN_SOURCE_MANUAL = "manual" OPEN_SOURCE_KEY_AUTO = "key_auto" OPEN_SOURCE_KEY_FIB = "key_fib" OPEN_SOURCE_KEY_TRIGGER = "key_trigger" OPEN_SOURCE_TREND = "trend" OPEN_SOURCE_ROLL = "roll" FULL_MARGIN_BLOCKED_SOURCES = frozenset( {OPEN_SOURCE_KEY_AUTO, OPEN_SOURCE_KEY_FIB, OPEN_SOURCE_TREND, OPEN_SOURCE_ROLL} ) def normalize_position_sizing_mode(raw: Optional[str]) -> str: v = (raw or MODE_RISK).strip().lower() if v in ("full", "full_margin", "fullmargin", "全仓", "全仓杠杆"): return MODE_FULL_MARGIN return MODE_RISK if v in ("risk", "r", "以损定仓", "") else MODE_RISK def load_position_sizing_mode(env: Optional[dict] = None) -> str: e = env if env is not None else os.environ return normalize_position_sizing_mode(e.get("POSITION_SIZING_MODE")) def is_full_margin_mode(mode: str) -> bool: return normalize_position_sizing_mode(mode) == MODE_FULL_MARGIN def mode_label_zh(mode: str) -> str: return "全仓杠杆" if is_full_margin_mode(mode) else "以损定仓" def leverage_for_full_margin(symbol: str, btc_leverage: int, alt_leverage: int) -> int: sym = (symbol or "").strip().upper() if sym.startswith("BTC") or sym.startswith("ETH"): return max(1, int(btc_leverage or 10)) return max(1, int(alt_leverage or 5)) def round_funds(value: float, decimals: int = 2) -> float: return round(float(value), int(decimals)) def risk_percent_for_storage(mode: str, risk_percent: float) -> Optional[float]: """全仓杠杆:库内不写风险百分比(仅 risk_amount U)。""" if is_full_margin_mode(mode): return None return risk_percent def format_risk_display_text( mode: str, risk_percent: Optional[float], risk_amount: Optional[float], *, decimals: int = 2, ) -> str: """持仓/通知「风险」文案:全仓仅 U;以损定仓为 %≈U。""" amt: Optional[float] = None if risk_amount is not None and risk_amount != "": try: amt = float(risk_amount) except (TypeError, ValueError): amt = None if is_full_margin_mode(mode): if amt is None: return "—" return f"{round_funds(amt, decimals)}U" pct: Optional[float] = None if risk_percent is not None and risk_percent != "": try: pct = float(risk_percent) except (TypeError, ValueError): pct = None pct_txt = f"{pct:g}" if pct is not None else "—" amt_txt = round_funds(amt, decimals) if amt is not None else "—" return f"{pct_txt}%≈{amt_txt}U" def assert_open_source_allowed(mode: str, source: str) -> Tuple[bool, str]: if not is_full_margin_mode(mode): return True, "" src = (source or "").strip().lower() if src in FULL_MARGIN_BLOCKED_SOURCES: return False, ( "当前为全仓杠杆模式(POSITION_SIZING_MODE=full_margin)," "不允许关键位突破/斐波自动开仓、趋势回调与顺势加仓;" "仅支持实盘人工下单与阻力/支撑提醒。" ) return True, "" def full_margin_requires_flat_position(active_count: int) -> Tuple[bool, str]: if active_count > 0: return False, "全仓杠杆模式仅允许单仓且无其它持仓,请先平仓后再开仓" return True, "" def compute_full_margin_sizing( *, symbol: str, available_usdt: float, capital_base: float, buffer_ratio: float, btc_leverage: int, alt_leverage: int, funds_decimals: int = 2, ) -> Tuple[Optional[dict[str, Any]], Optional[str]]: if available_usdt is None or float(available_usdt) <= 0: return None, "全仓杠杆:无法读取合约账户可用保证金" lev = leverage_for_full_margin(symbol, btc_leverage, alt_leverage) margin = round_funds(float(available_usdt) * float(buffer_ratio), funds_decimals) if margin <= 0: return None, "全仓杠杆:可用保证金不足" notional = round_funds(margin * lev, funds_decimals) ratio = round(margin / float(capital_base) * 100, 2) if capital_base else 0.0 return { "margin_capital": margin, "leverage": lev, "notional_value": notional, "position_ratio": ratio, "mode": MODE_FULL_MARGIN, }, None