feat: add fixed RR stop-loss mode for manual live orders on all instances

Co-authored-by: Cursor <cursoragent@cursor.com>

manual_sltp_lib.py

@@ -0,0 +1,136 @@
+"""实盘人工下单：止盈止损模式（价格 / 百分比 / 固定盈亏比）。"""
+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,
+    )