Files
crypto_monitor/position_sizing_lib.py
2026-06-14 00:42:21 +08:00

137 lines
4.6 KiB
Python

"""
四所共用:计仓模式 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