"""实盘人工下单:止盈止损模式(价格 / 百分比 / 固定盈亏比)。""" from __future__ import annotations from typing import Any, Optional, Tuple MANUAL_FIXED_RR_DEFAULT = 1.5 SLTP_MODE_PRICE = "price" SLTP_MODE_PCT = "pct" SLTP_MODE_FIXED_RR = "fixed_rr" OPEN_SLTP_MODES = frozenset({SLTP_MODE_PRICE, SLTP_MODE_PCT, SLTP_MODE_FIXED_RR}) ENTRUST_SLTP_MODES = frozenset({SLTP_MODE_PRICE, SLTP_MODE_PCT}) def normalize_open_sltp_mode(raw: Optional[str]) -> str: mode = (raw or SLTP_MODE_FIXED_RR).strip().lower() if mode in OPEN_SLTP_MODES: return mode return SLTP_MODE_PRICE def normalize_entrust_sltp_mode(raw: Optional[str]) -> str: mode = (raw or SLTP_MODE_PRICE).strip().lower() if mode in ENTRUST_SLTP_MODES: return mode return SLTP_MODE_PRICE def parse_fixed_rr(raw: Any, *, default: float = MANUAL_FIXED_RR_DEFAULT) -> float: try: v = float(raw) if v > 0: return v except (TypeError, ValueError): pass return float(default) def calc_tp_from_fixed_rr( direction: str, entry_price: float, stop_loss: float, rr_ratio: float, ) -> float: entry = float(entry_price) sl = float(stop_loss) rr = float(rr_ratio) if entry <= 0 or sl <= 0 or rr <= 0: raise ValueError("固定盈亏比参数无效") side = (direction or "long").strip().lower() if side == "short": risk = sl - entry if risk <= 0: raise ValueError("止损方向不合法:做空时止损须高于入场价") return entry - risk * rr risk = entry - sl if risk <= 0: raise ValueError("止损方向不合法:做多时止损须低于入场价") return entry + risk * rr def _resolve_pct_sltp(direction: str, live_price: float, data: dict[str, Any]) -> Tuple[float, float]: sl_pct = float(data.get("sl_pct") or 0) tp_pct = float(data.get("tp_pct") or 0) if sl_pct <= 0 or tp_pct <= 0: raise ValueError("百分比止盈止损须为正数") sl_ratio = sl_pct / 100.0 tp_ratio = tp_pct / 100.0 entry = float(live_price) if (direction or "long").strip().lower() == "short": stop_loss = entry * (1 + sl_ratio) take_profit = entry * (1 - tp_ratio) else: stop_loss = entry * (1 - sl_ratio) take_profit = entry * (1 + tp_ratio) return stop_loss, take_profit def _resolve_price_sltp( data: dict[str, Any], *, fallback_sl: Optional[float] = None, fallback_tp: Optional[float] = None, require_tp: bool = True, ) -> Tuple[float, float]: stop_loss = float(data.get("sl") or data.get("stop_loss") or 0) take_profit = float(data.get("tp") or data.get("take_profit") or data.get("tgt") or 0) if stop_loss <= 0 and fallback_sl is not None: stop_loss = float(fallback_sl) if take_profit <= 0 and fallback_tp is not None: take_profit = float(fallback_tp) if stop_loss <= 0: raise ValueError("止损价格须大于 0" if require_tp else "请填写止损价格") if require_tp and take_profit <= 0: raise ValueError("止盈止损价格须大于 0" if fallback_tp is None else "请填写止盈价格,或保留原计划止盈") return stop_loss, take_profit def resolve_open_sltp_prices( direction: str, live_price: float, sltp_mode: Optional[str], data: dict[str, Any], ) -> Tuple[float, float]: """新开仓 /add_order:支持 price、pct、fixed_rr。""" mode = normalize_open_sltp_mode(sltp_mode) if mode == SLTP_MODE_PCT: return _resolve_pct_sltp(direction, live_price, data) if mode == SLTP_MODE_FIXED_RR: stop_loss, _ = _resolve_price_sltp(data, require_tp=False) rr = parse_fixed_rr(data.get("fixed_rr")) take_profit = calc_tp_from_fixed_rr(direction, live_price, stop_loss, rr) return stop_loss, take_profit return _resolve_price_sltp(data, require_tp=True) def resolve_entrust_sltp_prices( direction: str, live_price: float, sltp_mode: Optional[str], data: dict[str, Any], *, fallback_sl: Optional[float] = None, fallback_tp: Optional[float] = None, ) -> Tuple[float, float]: """持仓委托弹窗:仅 price / pct,不校验盈亏比。""" mode = normalize_entrust_sltp_mode(sltp_mode) if mode == SLTP_MODE_PCT: return _resolve_pct_sltp(direction, live_price, data) return _resolve_price_sltp( data, fallback_sl=fallback_sl, fallback_tp=fallback_tp, require_tp=True, )