dekun
137 lines
4.5 KiB
Python
137 lines
4.5 KiB
Python
"""实盘人工下单：止盈止损模式（价格 / 百分比 / 固定盈亏比）。"""
|
|
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,
|
|
)
|